Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move backup backup onboarding API to an onboarding platform (#142713) 

* Move backup backup onboarding API to an onboarding platform

* Move additional test from onboarding to backup

* Remove backup tests from onboarding
pull/142785/head
Erik Montnemery 2025-04-12 09:41:54 +02:00 committed by GitHub
parent ad3c4d24b8
commit 234c4c1958
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 560 additions and 486 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
homeassistant/components/backup/onboarding.py Normal file
View File

@ -0,0 +1,143 @@
"""Backup onboarding views."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from functools import wraps
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, Concatenate
from aiohttp import web
from aiohttp.web_exceptions import HTTPUnauthorized
import voluptuous as vol
from homeassistant.components.http import KEY_HASS
from homeassistant.components.http.data_validator import RequestDataValidator
from homeassistant.components.onboarding import (
BaseOnboardingView,
NoAuthBaseOnboardingView,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.backup import async_get_manager as async_get_backup_manager
from . import BackupManager, Folder, IncorrectPasswordError, http as backup_http
if TYPE_CHECKING:
from homeassistant.components.onboarding import OnboardingStoreData
async def async_setup_views(hass: HomeAssistant, data: OnboardingStoreData) -> None:
"""Set up the backup views."""
hass.http.register_view(BackupInfoView(data))
hass.http.register_view(RestoreBackupView(data))
hass.http.register_view(UploadBackupView(data))
def with_backup_manager[_ViewT: BaseOnboardingView, **_P](
func: Callable[
Concatenate[_ViewT, BackupManager, 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 inject manager."""
@wraps(func)
async def with_backup(
self: _ViewT,
request: web.Request,
*args: _P.args,
**kwargs: _P.kwargs,
) -> web.Response:
"""Check admin and call function."""
if self._data["done"]:
raise HTTPUnauthorized
try:
manager = await async_get_backup_manager(request.app[KEY_HASS])
except HomeAssistantError:
return self.json(
{"code": "backup_disabled"},
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
)
return await func(self, manager, request, *args, **kwargs)
return with_backup
class BackupInfoView(NoAuthBaseOnboardingView):
"""Get backup info view."""
url = "/api/onboarding/backup/info"
name = "api:onboarding:backup:info"
@with_backup_manager
async def get(self, manager: BackupManager, request: web.Request) -> web.Response:
"""Return backup info."""
backups, _ = await manager.async_get_backups()
return self.json(
{
"backups": list(backups.values()),
"state": manager.state,
"last_action_event": manager.last_action_event,
}
)
class RestoreBackupView(NoAuthBaseOnboardingView):
"""Restore backup view."""
url = "/api/onboarding/backup/restore"
name = "api:onboarding:backup:restore"
@RequestDataValidator(
vol.Schema(
{
vol.Required("backup_id"): str,
vol.Required("agent_id"): str,
vol.Optional("password"): str,
vol.Optional("restore_addons"): [str],
vol.Optional("restore_database", default=True): bool,
vol.Optional("restore_folders"): [vol.Coerce(Folder)],
}
)
)
@with_backup_manager
async def post(
self, manager: BackupManager, request: web.Request, data: dict[str, Any]
) -> web.Response:
"""Restore a backup."""
try:
await manager.async_restore_backup(
data["backup_id"],
agent_id=data["agent_id"],
password=data.get("password"),
restore_addons=data.get("restore_addons"),
restore_database=data["restore_database"],
restore_folders=data.get("restore_folders"),
restore_homeassistant=True,
)
except IncorrectPasswordError:
return self.json(
{"code": "incorrect_password"}, status_code=HTTPStatus.BAD_REQUEST
)
except HomeAssistantError as err:
return self.json(
{"code": "restore_failed", "message": str(err)},
status_code=HTTPStatus.BAD_REQUEST,
)
return web.Response(status=HTTPStatus.OK)
class UploadBackupView(NoAuthBaseOnboardingView, backup_http.UploadBackupView):
"""Upload backup view."""
url = "/api/onboarding/backup/upload"
name = "api:onboarding:backup:upload"
@with_backup_manager
async def post(self, manager: BackupManager, request: web.Request) -> web.Response:
"""Upload a backup file."""
return await self._post(request)

123
homeassistant/components/onboarding/views.py
View File

@ -3,11 +3,9 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine
from functools import wraps
from http import HTTPStatus
import logging
from typing import TYPE_CHECKING, Any, Concatenate, Protocol, cast
from typing import TYPE_CHECKING, Any, Protocol, cast
from aiohttp import web
from aiohttp.web_exceptions import HTTPUnauthorized
@ -17,19 +15,11 @@ from homeassistant.auth.const import GROUP_ID_ADMIN
from homeassistant.auth.providers.homeassistant import HassAuthProvider
from homeassistant.components import person
from homeassistant.components.auth import indieauth
from homeassistant.components.backup import (
BackupManager,
Folder,
IncorrectPasswordError,
http as backup_http,
)
from homeassistant.components.http import KEY_HASS, KEY_HASS_REFRESH_TOKEN_ID
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, 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 async_setup_component, async_wait_component
@ -61,9 +51,6 @@ async def async_setup(
hass.http.register_view(CoreConfigOnboardingView(data, store))
hass.http.register_view(IntegrationOnboardingView(data, store))
hass.http.register_view(AnalyticsOnboardingView(data, store))
hass.http.register_view(BackupInfoView(data))
hass.http.register_view(RestoreBackupView(data))
hass.http.register_view(UploadBackupView(data))
hass.http.register_view(WaitIntegrationOnboardingView(data))
@ -377,114 +364,6 @@ class AnalyticsOnboardingView(_BaseOnboardingStepView):
return self.json({})
def with_backup_manager[_ViewT: BaseOnboardingView, **_P](
func: Callable[
Concatenate[_ViewT, BackupManager, 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 inject manager."""
@wraps(func)
async def with_backup(
self: _ViewT,
request: web.Request,
*args: _P.args,
**kwargs: _P.kwargs,
) -> web.Response:
"""Check admin and call function."""
if self._data["done"]:
raise HTTPUnauthorized
try:
manager = await async_get_backup_manager(request.app[KEY_HASS])
except HomeAssistantError:
return self.json(
{"code": "backup_disabled"},
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
)
return await func(self, manager, request, *args, **kwargs)
return with_backup
class BackupInfoView(NoAuthBaseOnboardingView):
"""Get backup info view."""
url = "/api/onboarding/backup/info"
name = "api:onboarding:backup:info"
@with_backup_manager
async def get(self, manager: BackupManager, request: web.Request) -> web.Response:
"""Return backup info."""
backups, _ = await manager.async_get_backups()
return self.json(
{
"backups": list(backups.values()),
"state": manager.state,
"last_action_event": manager.last_action_event,
}
)
class RestoreBackupView(NoAuthBaseOnboardingView):
"""Restore backup view."""
url = "/api/onboarding/backup/restore"
name = "api:onboarding:backup:restore"
@RequestDataValidator(
vol.Schema(
{
vol.Required("backup_id"): str,
vol.Required("agent_id"): str,
vol.Optional("password"): str,
vol.Optional("restore_addons"): [str],
vol.Optional("restore_database", default=True): bool,
vol.Optional("restore_folders"): [vol.Coerce(Folder)],
}
)
)
@with_backup_manager
async def post(
self, manager: BackupManager, request: web.Request, data: dict[str, Any]
) -> web.Response:
"""Restore a backup."""
try:
await manager.async_restore_backup(
data["backup_id"],
agent_id=data["agent_id"],
password=data.get("password"),
restore_addons=data.get("restore_addons"),
restore_database=data["restore_database"],
restore_folders=data.get("restore_folders"),
restore_homeassistant=True,
)
except IncorrectPasswordError:
return self.json(
{"code": "incorrect_password"}, status_code=HTTPStatus.BAD_REQUEST
)
except HomeAssistantError as err:
return self.json(
{"code": "restore_failed", "message": str(err)},
status_code=HTTPStatus.BAD_REQUEST,
)
return web.Response(status=HTTPStatus.OK)
class UploadBackupView(NoAuthBaseOnboardingView, backup_http.UploadBackupView):
"""Upload backup view."""
url = "/api/onboarding/backup/upload"
name = "api:onboarding:backup:upload"
@with_backup_manager
async def post(self, manager: BackupManager, request: web.Request) -> web.Response:
"""Upload a backup file."""
return await self._post(request)
@callback
def _async_get_hass_provider(hass: HomeAssistant) -> HassAuthProvider:
"""Get the Home Assistant auth provider."""

4
script/hassfest/dependencies.py
View File

@ -173,10 +173,6 @@ IGNORE_VIOLATIONS = {
"logbook",
# Temporary needed for migration until 2024.10
("conversation", "assist_pipeline"),
# The onboarding integration provides limited backup for use
# during onboarding. The onboarding integration waits for the backup manager
# and to be ready before calling any backup functionality.
("onboarding", "backup"),
}

0
tests/components/onboarding/snapshots/test_views.ambr → tests/components/backup/snapshots/test_onboarding.ambr
View File

414
tests/components/backup/test_onboarding.py Normal file
View File

@ -0,0 +1,414 @@
"""Test the onboarding views."""
from io import StringIO
from typing import Any
from unittest.mock import ANY, patch
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components import backup, onboarding
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.backup import async_initialize_backup
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)
def auth_active(hass: HomeAssistant) -> None:
"""Ensure auth is always active."""
hass.loop.run_until_complete(
register_auth_provider(hass, {"type": "homeassistant"})
)
@pytest.mark.parametrize(
("method", "view", "kwargs"),
[
("get", "backup/info", {}),
(
"post",
"backup/restore",
{"json": {"backup_id": "abc123", "agent_id": "test"}},
),
("post", "backup/upload", {}),
],
)
async def test_onboarding_view_after_done(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
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", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
resp = await client.request(method, f"/api/onboarding/{view}", **kwargs)
assert resp.status == 401
@pytest.mark.parametrize(
("method", "view", "kwargs"),
[
("get", "backup/info", {}),
(
"post",
"backup/restore",
{"json": {"backup_id": "abc123", "agent_id": "test"}},
),
("post", "backup/upload", {}),
],
)
async def test_onboarding_backup_view_without_backup(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
method: str,
view: str,
kwargs: dict[str, Any],
) -> None:
"""Test interacting with backup wievs when backup integration is missing."""
mock_onboarding_storage(hass_storage, {"done": []})
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 == 404
async def test_onboarding_backup_info(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test backup info."""
mock_onboarding_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
backups = {
"abc123": backup.ManagerBackup(
addons=[backup.AddonInfo(name="Test", slug="test", version="1.0.0")],
agents={
"backup.local": backup.manager.AgentBackupStatus(protected=True, size=0)
},
backup_id="abc123",
date="1970-01-01T00:00:00.000Z",
database_included=True,
extra_metadata={"instance_id": "abc123", "with_automatic_settings": True},
folders=[backup.Folder.MEDIA, backup.Folder.SHARE],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Test",
failed_agent_ids=[],
with_automatic_settings=True,
),
"def456": backup.ManagerBackup(
addons=[],
agents={
"test.remote": backup.manager.AgentBackupStatus(protected=True, size=0)
},
backup_id="def456",
date="1980-01-01T00:00:00.000Z",
database_included=False,
extra_metadata={
"instance_id": "unknown_uuid",
"with_automatic_settings": True,
},
folders=[backup.Folder.MEDIA, backup.Folder.SHARE],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Test 2",
failed_agent_ids=[],
with_automatic_settings=None,
),
}
with patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backups",
return_value=(backups, {}),
):
resp = await client.get("/api/onboarding/backup/info")
assert resp.status == 200
assert await resp.json() == snapshot
@pytest.mark.parametrize(
("params", "expected_kwargs"),
[
(
{"backup_id": "abc123", "agent_id": "backup.local"},
{
"agent_id": "backup.local",
"password": None,
"restore_addons": None,
"restore_database": True,
"restore_folders": None,
"restore_homeassistant": True,
},
),
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1"],
"restore_database": True,
"restore_folders": ["media"],
},
{
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1"],
"restore_database": True,
"restore_folders": [backup.Folder.MEDIA],
"restore_homeassistant": True,
},
),
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1", "addon_2"],
"restore_database": False,
"restore_folders": ["media", "share"],
},
{
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1", "addon_2"],
"restore_database": False,
"restore_folders": [backup.Folder.MEDIA, backup.Folder.SHARE],
"restore_homeassistant": True,
},
),
],
)
async def test_onboarding_backup_restore(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
expected_kwargs: dict[str, Any],
) -> None:
"""Test restore backup."""
mock_onboarding_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == 200
mock_restore.assert_called_once_with("abc123", **expected_kwargs)
@pytest.mark.parametrize(
("params", "restore_error", "expected_status", "expected_json", "restore_calls"),
[
# Missing agent_id
(
{"backup_id": "abc123"},
None,
400,
{
"message": "Message format incorrect: required key not provided @ data['agent_id']"
},
0,
),
# Missing backup_id
(
{"agent_id": "backup.local"},
None,
400,
{
"message": "Message format incorrect: required key not provided @ data['backup_id']"
},
0,
),
# Invalid restore_database
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"restore_database": "yes_please",
},
None,
400,
{
"message": "Message format incorrect: expected bool for dictionary value @ data['restore_database']"
},
0,
),
# Invalid folder
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"restore_folders": ["invalid"],
},
None,
400,
{
"message": "Message format incorrect: expected Folder or one of 'share', 'addons/local', 'ssl', 'media' @ data['restore_folders'][0]"
},
0,
),
# Wrong password
(
{"backup_id": "abc123", "agent_id": "backup.local"},
backup.IncorrectPasswordError,
400,
{"code": "incorrect_password"},
1,
),
# Home Assistant error
(
{"backup_id": "abc123", "agent_id": "backup.local"},
HomeAssistantError("Boom!"),
400,
{"code": "restore_failed", "message": "Boom!"},
1,
),
],
)
async def test_onboarding_backup_restore_error(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
restore_error: Exception | None,
expected_status: int,
expected_json: str,
restore_calls: int,
) -> None:
"""Test restore backup fails."""
mock_onboarding_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
side_effect=restore_error,
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == expected_status
assert await resp.json() == expected_json
assert len(mock_restore.mock_calls) == restore_calls
@pytest.mark.parametrize(
("params", "restore_error", "expected_status", "expected_message", "restore_calls"),
[
# Unexpected error
(
{"backup_id": "abc123", "agent_id": "backup.local"},
Exception("Boom!"),
500,
"500 Internal Server Error",
1,
),
],
)
async def test_onboarding_backup_restore_unexpected_error(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
restore_error: Exception | None,
expected_status: int,
expected_message: str,
restore_calls: int,
) -> None:
"""Test restore backup fails."""
mock_onboarding_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
side_effect=restore_error,
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == expected_status
assert (await resp.content.read()).decode().startswith(expected_message)
assert len(mock_restore.mock_calls) == restore_calls
async def test_onboarding_backup_upload(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
) -> None:
"""Test upload backup."""
mock_onboarding_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_receive_backup",
return_value="abc123",
) as mock_receive:
resp = await client.post(
"/api/onboarding/backup/upload?agent_id=backup.local",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert await resp.json() == {"backup_id": "abc123"}
mock_receive.assert_called_once_with(agent_ids=["backup.local"], contents=ANY)

362
tests/components/onboarding/test_views.py
View File

@ -3,20 +3,16 @@
import asyncio
from collections.abc import AsyncGenerator
from http import HTTPStatus
from io import StringIO
import os
from typing import Any
from unittest.mock import ANY, AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock, patch
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components import backup, onboarding
from homeassistant.components import onboarding
from homeassistant.components.onboarding import const, views
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import area_registry as ar
from homeassistant.helpers.backup import async_initialize_backup
from homeassistant.setup import async_set_domains_to_be_loaded, async_setup_component
from . import mock_storage
@ -632,13 +628,6 @@ async def test_onboarding_installation_type(
("method", "view", "kwargs"),
[
("get", "installation_type", {}),
("get", "backup/info", {}),
(
"post",
"backup/restore",
{"json": {"backup_id": "abc123", "agent_id": "test"}},
),
("post", "backup/upload", {}),
],
)
async def test_onboarding_view_after_done(
@ -723,353 +712,6 @@ async def test_complete_onboarding(
listener_3.assert_called_once_with()
@pytest.mark.parametrize(
("method", "view", "kwargs"),
[
("get", "backup/info", {}),
(
"post",
"backup/restore",
{"json": {"backup_id": "abc123", "agent_id": "test"}},
),
("post", "backup/upload", {}),
],
)
async def test_onboarding_backup_view_without_backup(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
method: str,
view: str,
kwargs: dict[str, Any],
) -> None:
"""Test interacting with backup wievs when backup integration is missing."""
mock_storage(hass_storage, {"done": []})
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 == 500
assert await resp.json() == {"code": "backup_disabled"}
async def test_onboarding_backup_info(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
snapshot: SnapshotAssertion,
) -> None:
"""Test backup info."""
mock_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
backups = {
"abc123": backup.ManagerBackup(
addons=[backup.AddonInfo(name="Test", slug="test", version="1.0.0")],
agents={
"backup.local": backup.manager.AgentBackupStatus(protected=True, size=0)
},
backup_id="abc123",
date="1970-01-01T00:00:00.000Z",
database_included=True,
extra_metadata={"instance_id": "abc123", "with_automatic_settings": True},
folders=[backup.Folder.MEDIA, backup.Folder.SHARE],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Test",
failed_agent_ids=[],
with_automatic_settings=True,
),
"def456": backup.ManagerBackup(
addons=[],
agents={
"test.remote": backup.manager.AgentBackupStatus(protected=True, size=0)
},
backup_id="def456",
date="1980-01-01T00:00:00.000Z",
database_included=False,
extra_metadata={
"instance_id": "unknown_uuid",
"with_automatic_settings": True,
},
folders=[backup.Folder.MEDIA, backup.Folder.SHARE],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Test 2",
failed_agent_ids=[],
with_automatic_settings=None,
),
}
with patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backups",
return_value=(backups, {}),
):
resp = await client.get("/api/onboarding/backup/info")
assert resp.status == 200
assert await resp.json() == snapshot
@pytest.mark.parametrize(
("params", "expected_kwargs"),
[
(
{"backup_id": "abc123", "agent_id": "backup.local"},
{
"agent_id": "backup.local",
"password": None,
"restore_addons": None,
"restore_database": True,
"restore_folders": None,
"restore_homeassistant": True,
},
),
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1"],
"restore_database": True,
"restore_folders": ["media"],
},
{
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1"],
"restore_database": True,
"restore_folders": [backup.Folder.MEDIA],
"restore_homeassistant": True,
},
),
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1", "addon_2"],
"restore_database": False,
"restore_folders": ["media", "share"],
},
{
"agent_id": "backup.local",
"password": "hunter2",
"restore_addons": ["addon_1", "addon_2"],
"restore_database": False,
"restore_folders": [backup.Folder.MEDIA, backup.Folder.SHARE],
"restore_homeassistant": True,
},
),
],
)
async def test_onboarding_backup_restore(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
expected_kwargs: dict[str, Any],
) -> None:
"""Test restore backup."""
mock_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == 200
mock_restore.assert_called_once_with("abc123", **expected_kwargs)
@pytest.mark.parametrize(
("params", "restore_error", "expected_status", "expected_json", "restore_calls"),
[
# Missing agent_id
(
{"backup_id": "abc123"},
None,
400,
{
"message": "Message format incorrect: required key not provided @ data['agent_id']"
},
0,
),
# Missing backup_id
(
{"agent_id": "backup.local"},
None,
400,
{
"message": "Message format incorrect: required key not provided @ data['backup_id']"
},
0,
),
# Invalid restore_database
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"restore_database": "yes_please",
},
None,
400,
{
"message": "Message format incorrect: expected bool for dictionary value @ data['restore_database']"
},
0,
),
# Invalid folder
(
{
"backup_id": "abc123",
"agent_id": "backup.local",
"restore_folders": ["invalid"],
},
None,
400,
{
"message": "Message format incorrect: expected Folder or one of 'share', 'addons/local', 'ssl', 'media' @ data['restore_folders'][0]"
},
0,
),
# Wrong password
(
{"backup_id": "abc123", "agent_id": "backup.local"},
backup.IncorrectPasswordError,
400,
{"code": "incorrect_password"},
1,
),
# Home Assistant error
(
{"backup_id": "abc123", "agent_id": "backup.local"},
HomeAssistantError("Boom!"),
400,
{"code": "restore_failed", "message": "Boom!"},
1,
),
],
)
async def test_onboarding_backup_restore_error(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
restore_error: Exception | None,
expected_status: int,
expected_json: str,
restore_calls: int,
) -> None:
"""Test restore backup fails."""
mock_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
side_effect=restore_error,
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == expected_status
assert await resp.json() == expected_json
assert len(mock_restore.mock_calls) == restore_calls
@pytest.mark.parametrize(
("params", "restore_error", "expected_status", "expected_message", "restore_calls"),
[
# Unexpected error
(
{"backup_id": "abc123", "agent_id": "backup.local"},
Exception("Boom!"),
500,
"500 Internal Server Error",
1,
),
],
)
async def test_onboarding_backup_restore_unexpected_error(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
params: dict[str, Any],
restore_error: Exception | None,
expected_status: int,
expected_message: str,
restore_calls: int,
) -> None:
"""Test restore backup fails."""
mock_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
side_effect=restore_error,
) as mock_restore:
resp = await client.post("/api/onboarding/backup/restore", json=params)
assert resp.status == expected_status
assert (await resp.content.read()).decode().startswith(expected_message)
assert len(mock_restore.mock_calls) == restore_calls
async def test_onboarding_backup_upload(
hass: HomeAssistant,
hass_storage: dict[str, Any],
hass_client: ClientSessionGenerator,
) -> None:
"""Test upload backup."""
mock_storage(hass_storage, {"done": []})
assert await async_setup_component(hass, "onboarding", {})
async_initialize_backup(hass)
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_client()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_receive_backup",
return_value="abc123",
) as mock_receive:
resp = await client.post(
"/api/onboarding/backup/upload?agent_id=backup.local",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert await resp.json() == {"backup_id": "abc123"}
mock_receive.assert_called_once_with(agent_ids=["backup.local"], contents=ANY)
@pytest.mark.parametrize(
("domain", "expected_result"),
[