jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-18 18:57:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update hassio to use the backup integration to make backups before update (#136235)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Erik Montnemery 2025-01-27 08:25:22 +01:00 committed by GitHub
parent 69938545df
commit 245ee2498e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 979 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
homeassistant/components/backup/__init__.py
View File

@ -1,6 +1,7 @@
"""The Backup integration."""
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.hassio import is_hassio
from homeassistant.helpers.typing import ConfigType
@ -19,6 +20,7 @@ from .const import DATA_MANAGER, DOMAIN
from .http import async_register_http_views
from .manager import (
BackupManager,
BackupManagerError,
BackupPlatformProtocol,
BackupReaderWriter,
BackupReaderWriterError,
@ -39,6 +41,7 @@ __all__ = [
"BackupAgent",
"BackupAgentError",
"BackupAgentPlatformProtocol",
"BackupManagerError",
"BackupPlatformProtocol",
"BackupReaderWriter",
"BackupReaderWriterError",
@ -90,18 +93,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_handle_create_automatic_service(call: ServiceCall) -> None:
"""Service handler for creating automatic backups."""
config_data = backup_manager.config.data
await backup_manager.async_create_backup(
agent_ids=config_data.create_backup.agent_ids,
include_addons=config_data.create_backup.include_addons,
include_all_addons=config_data.create_backup.include_all_addons,
include_database=config_data.create_backup.include_database,
include_folders=config_data.create_backup.include_folders,
include_homeassistant=True, # always include HA
name=config_data.create_backup.name,
password=config_data.create_backup.password,
with_automatic_settings=True,
)
await backup_manager.async_create_automatic_backup()
if not with_hassio:
hass.services.async_register(DOMAIN, "create", async_handle_create_service)
@ -112,3 +104,15 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async_register_http_views(hass)
return True
@callback
def async_get_manager(hass: HomeAssistant) -> BackupManager:
"""Get the backup manager instance.
Raises HomeAssistantError if the backup integration is not available.
"""
if DATA_MANAGER not in hass.data:
raise HomeAssistantError("Backup integration is not available")
return hass.data[DATA_MANAGER]

13
homeassistant/components/backup/config.py
View File

@ -390,22 +390,11 @@ class BackupSchedule:
async def _create_backup(now: datetime) -> None:
"""Create backup."""
manager.remove_next_backup_event = None
config_data = manager.config.data
self._schedule_next(cron_pattern, manager)
# create the backup
try:
await manager.async_create_backup(
agent_ids=config_data.create_backup.agent_ids,
include_addons=config_data.create_backup.include_addons,
include_all_addons=config_data.create_backup.include_all_addons,
include_database=config_data.create_backup.include_database,
include_folders=config_data.create_backup.include_folders,
include_homeassistant=True, # always include HA
name=config_data.create_backup.name,
password=config_data.create_backup.password,
with_automatic_settings=True,
)
await manager.async_create_automatic_backup()
except BackupManagerError as err:
LOGGER.error("Error creating backup: %s", err)
except Exception: # noqa: BLE001

15
homeassistant/components/backup/manager.py
View File

@ -698,6 +698,21 @@ class BackupManager:
await self._backup_finish_task
return new_backup
async def async_create_automatic_backup(self) -> NewBackup:
"""Create a backup with automatic backup settings."""
config_data = self.config.data
return await self.async_create_backup(
agent_ids=config_data.create_backup.agent_ids,
include_addons=config_data.create_backup.include_addons,
include_all_addons=config_data.create_backup.include_all_addons,
include_database=config_data.create_backup.include_database,
include_folders=config_data.create_backup.include_folders,
include_homeassistant=True, # always include HA
name=config_data.create_backup.name,
password=config_data.create_backup.password,
with_automatic_settings=True,
)
async def async_initiate_backup(
self,
*,

67
homeassistant/components/hassio/backup.py
View File

@ -8,6 +8,7 @@ import logging
from pathlib import Path
from typing import Any, cast
from aiohasupervisor import SupervisorClient
from aiohasupervisor.exceptions import (
SupervisorBadRequestError,
SupervisorError,
@ -23,6 +24,7 @@ from homeassistant.components.backup import (
AddonInfo,
AgentBackup,
BackupAgent,
BackupManagerError,
BackupReaderWriter,
BackupReaderWriterError,
CreateBackupEvent,
@ -31,7 +33,9 @@ from homeassistant.components.backup import (
NewBackup,
RestoreBackupEvent,
WrittenBackup,
async_get_manager as async_get_backup_manager,
)
from homeassistant.const import __version__ as HAVERSION
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -477,3 +481,66 @@ class SupervisorBackupReaderWriter(BackupReaderWriter):
self._hass, EVENT_SUPERVISOR_EVENT, handle_signal
)
return unsub
async def _default_agent(client: SupervisorClient) -> str:
"""Return the default agent for creating a backup."""
mounts = await client.mounts.info()
default_mount = mounts.default_backup_mount
return f"hassio.{default_mount if default_mount is not None else 'local'}"
async def backup_addon_before_update(
hass: HomeAssistant,
addon: str,
addon_name: str | None,
installed_version: str | None,
) -> None:
"""Prepare for updating an add-on."""
backup_manager = hass.data[DATA_MANAGER]
client = get_supervisor_client(hass)
# Use the password from automatic settings if available
if backup_manager.config.data.create_backup.agent_ids:
password = backup_manager.config.data.create_backup.password
else:
password = None
try:
await backup_manager.async_create_backup(
agent_ids=[await _default_agent(client)],
include_addons=[addon],
include_all_addons=False,
include_database=False,
include_folders=None,
include_homeassistant=False,
name=f"{addon_name or addon} {installed_version or '<unknown>'}",
password=password,
)
except BackupManagerError as err:
raise HomeAssistantError(f"Error creating backup: {err}") from err
async def backup_core_before_update(hass: HomeAssistant) -> None:
"""Prepare for updating core."""
backup_manager = async_get_backup_manager(hass)
client = get_supervisor_client(hass)
try:
if backup_manager.config.data.create_backup.agent_ids:
# Create a backup with automatic settings
await backup_manager.async_create_automatic_backup()
else:
# Create a manual backup
await backup_manager.async_create_backup(
agent_ids=[await _default_agent(client)],
include_addons=None,
include_all_addons=False,
include_database=True,
include_folders=None,
include_homeassistant=True,
name=f"Home Assistant Core {HAVERSION}",
password=None,
)
except BackupManagerError as err:
raise HomeAssistantError(f"Error creating backup: {err}") from err

35
homeassistant/components/hassio/update.py
View File

@ -4,12 +4,8 @@ from __future__ import annotations
from typing import Any
from aiohasupervisor import SupervisorError
from aiohasupervisor.models import (
HomeAssistantUpdateOptions,
OSUpdate,
StoreAddonUpdate,
)
from aiohasupervisor import SupervisorClient, SupervisorError
from aiohasupervisor.models import OSUpdate
from awesomeversion import AwesomeVersion, AwesomeVersionStrategy
from homeassistant.components.update import (
@ -40,6 +36,7 @@ from .entity import (
HassioOSEntity,
HassioSupervisorEntity,
)
from .update_helper import update_addon, update_core
ENTITY_DESCRIPTION = UpdateEntityDescription(
name="Update",
@ -163,13 +160,9 @@ class SupervisorAddonUpdateEntity(HassioAddonEntity, UpdateEntity):
**kwargs: Any,
) -> None:
"""Install an update."""
try:
await self.coordinator.supervisor_client.store.update_addon(
self._addon_slug, StoreAddonUpdate(backup=backup)
)
except SupervisorError as err:
raise HomeAssistantError(f"Error updating {self.title}: {err}") from err
await update_addon(
self.hass, self._addon_slug, backup, self.title, self.installed_version
)
await self.coordinator.force_info_update_supervisor()
@ -303,11 +296,11 @@ class SupervisorCoreUpdateEntity(HassioCoreEntity, UpdateEntity):
self, version: str | None, backup: bool, **kwargs: Any
) -> None:
"""Install an update."""
try:
await self.coordinator.supervisor_client.homeassistant.update(
HomeAssistantUpdateOptions(version=version, backup=backup)
)
except SupervisorError as err:
raise HomeAssistantError(
f"Error updating Home Assistant Core: {err}"
) from err
await update_core(self.hass, version, backup)
async def _default_agent(client: SupervisorClient) -> str:
"""Return the default agent for creating a backup."""
mounts = await client.mounts.info()
default_mount = mounts.default_backup_mount
return f"hassio.{default_mount if default_mount is not None else 'local'}"

59
homeassistant/components/hassio/update_helper.py Normal file
View File

@ -0,0 +1,59 @@
"""Update helpers for Supervisor."""
from __future__ import annotations
from aiohasupervisor import SupervisorError
from aiohasupervisor.models import HomeAssistantUpdateOptions, StoreAddonUpdate
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .handler import get_supervisor_client
async def update_addon(
hass: HomeAssistant,
addon: str,
backup: bool,
addon_name: str | None,
installed_version: str | None,
) -> None:
"""Update an addon.
Optionally make a backup before updating.
"""
client = get_supervisor_client(hass)
if backup:
# pylint: disable-next=import-outside-toplevel
from .backup import backup_addon_before_update
await backup_addon_before_update(hass, addon, addon_name, installed_version)
try:
await client.store.update_addon(addon, StoreAddonUpdate(backup=False))
except SupervisorError as err:
raise HomeAssistantError(
f"Error updating {addon_name or addon}: {err}"
) from err
async def update_core(hass: HomeAssistant, version: str | None, backup: bool) -> None:
"""Update core.
Optionally make a backup before updating.
"""
client = get_supervisor_client(hass)
if backup:
# pylint: disable-next=import-outside-toplevel
from .backup import backup_core_before_update
await backup_core_before_update(hass)
try:
await client.homeassistant.update(
HomeAssistantUpdateOptions(version=version, backup=False)
)
except SupervisorError as err:
raise HomeAssistantError(f"Error updating Home Assistant Core: {err}") from err

48
homeassistant/components/hassio/websocket_api.py
View File

@ -9,6 +9,7 @@ import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.components.websocket_api import ActiveConnection
from homeassistant.const import ATTR_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import Unauthorized
import homeassistant.helpers.config_validation as cv
@ -23,7 +24,9 @@ from .const import (
ATTR_ENDPOINT,
ATTR_METHOD,
ATTR_SESSION_DATA_USER_ID,
ATTR_SLUG,
ATTR_TIMEOUT,
ATTR_VERSION,
ATTR_WS_EVENT,
DATA_COMPONENT,
EVENT_SUPERVISOR_EVENT,
@ -33,6 +36,8 @@ from .const import (
WS_TYPE_EVENT,
WS_TYPE_SUBSCRIBE,
)
from .coordinator import get_supervisor_info
from .update_helper import update_addon, update_core
SCHEMA_WEBSOCKET_EVENT = vol.Schema(
{vol.Required(ATTR_WS_EVENT): cv.string},
@ -58,6 +63,8 @@ def async_load_websocket_api(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, websocket_supervisor_event)
websocket_api.async_register_command(hass, websocket_supervisor_api)
websocket_api.async_register_command(hass, websocket_subscribe)
websocket_api.async_register_command(hass, websocket_update_addon)
websocket_api.async_register_command(hass, websocket_update_core)
@callback
@ -137,3 +144,44 @@ async def websocket_supervisor_api(
)
else:
connection.send_result(msg[WS_ID], result.get(ATTR_DATA, {}))
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(WS_TYPE): "hassio/update/addon",
vol.Required("addon"): str,
vol.Required("backup"): bool,
}
)
@websocket_api.async_response
async def websocket_update_addon(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Websocket handler to update an addon."""
addon_name: str | None = None
addon_version: str | None = None
addons: list = (get_supervisor_info(hass) or {}).get("addons", [])
for addon in addons:
if addon[ATTR_SLUG] == msg["addon"]:
addon_name = addon[ATTR_NAME]
addon_version = addon[ATTR_VERSION]
break
await update_addon(hass, msg["addon"], msg["backup"], addon_name, addon_version)
connection.send_result(msg[WS_ID])
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(WS_TYPE): "hassio/update/core",
vol.Required("backup"): bool,
}
)
@websocket_api.async_response
async def websocket_update_core(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
) -> None:
"""Websocket handler to update an addon."""
await update_core(hass, None, msg["backup"])
connection.send_result(msg[WS_ID])

6
tests/components/conftest.py
View File

@ -528,7 +528,7 @@ def resolution_suggestions_for_issue_fixture(supervisor_client: AsyncMock) -> As
@pytest.fixture(name="supervisor_client")
def supervisor_client() -> Generator[AsyncMock]:
"""Mock the supervisor client."""
mounts_info_mock = AsyncMock(spec_set=["mounts"])
mounts_info_mock = AsyncMock(spec_set=["default_backup_mount", "mounts"])
mounts_info_mock.mounts = []
supervisor_client = AsyncMock()
supervisor_client.addons = AsyncMock()
@ -572,6 +572,10 @@ def supervisor_client() -> Generator[AsyncMock]:
"homeassistant.components.hassio.repairs.get_supervisor_client",
return_value=supervisor_client,
),
patch(
"homeassistant.components.hassio.update_helper.get_supervisor_client",
return_value=supervisor_client,
),
):
yield supervisor_client

318
tests/components/hassio/test_update.py
View File

@ -2,14 +2,17 @@
from datetime import timedelta
import os
from typing import Any
from unittest.mock import AsyncMock, patch
from aiohasupervisor import SupervisorBadRequestError, SupervisorError
from aiohasupervisor.models import StoreAddonUpdate
from aiohasupervisor.models import HomeAssistantUpdateOptions, StoreAddonUpdate
import pytest
from homeassistant.components.backup import BackupManagerError
from homeassistant.components.hassio import DOMAIN
from homeassistant.components.hassio.const import REQUEST_REFRESH_DELAY
from homeassistant.const import __version__ as HAVERSION
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
@ -216,12 +219,119 @@ async def test_update_addon(hass: HomeAssistant, update_addon: AsyncMock) -> Non
assert result
await hass.async_block_till_done()
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.test_update"},
blocking=True,
)
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.test_update"},
blocking=True,
)
mock_create_backup.assert_not_called()
update_addon.assert_called_once_with("test", StoreAddonUpdate(backup=False))
@pytest.mark.parametrize(
("commands", "default_mount", "expected_kwargs"),
[
(
[],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": None,
},
),
(
[],
"my_nas",
{
"agent_ids": ["hassio.my_nas"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": None,
},
),
(
[
{
"type": "backup/config/update",
"create_backup": {
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"name": "cool_backup",
"password": "hunter2",
},
},
],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": "hunter2",
},
),
],
)
async def test_update_addon_with_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
update_addon: AsyncMock,
commands: list[dict[str, Any]],
default_mount: str | None,
expected_kwargs: dict[str, Any],
) -> None:
"""Test updating addon update entity."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_ws_client(hass)
for command in commands:
await client.send_json_auto_id(command)
result = await client.receive_json()
assert result["success"]
supervisor_client.mounts.info.return_value.default_backup_mount = default_mount
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.test_update", "backup": True},
blocking=True,
)
mock_create_backup.assert_called_once_with(**expected_kwargs)
update_addon.assert_called_once_with("test", StoreAddonUpdate(backup=False))
@ -264,13 +374,125 @@ async def test_update_core(hass: HomeAssistant, supervisor_client: AsyncMock) ->
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.home_assistant_core_update"},
blocking=True,
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.home_assistant_core_update"},
blocking=True,
)
mock_create_backup.assert_not_called()
supervisor_client.homeassistant.update.assert_called_once_with(
HomeAssistantUpdateOptions(version=None, backup=False)
)
@pytest.mark.parametrize(
("commands", "default_mount", "expected_kwargs"),
[
(
[],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": None,
"include_all_addons": False,
"include_database": True,
"include_folders": None,
"include_homeassistant": True,
"name": f"Home Assistant Core {HAVERSION}",
"password": None,
},
),
(
[],
"my_nas",
{
"agent_ids": ["hassio.my_nas"],
"include_addons": None,
"include_all_addons": False,
"include_database": True,
"include_folders": None,
"include_homeassistant": True,
"name": f"Home Assistant Core {HAVERSION}",
"password": None,
},
),
(
[
{
"type": "backup/config/update",
"create_backup": {
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"name": "cool_backup",
"password": "hunter2",
},
},
],
None,
{
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"include_homeassistant": True,
"name": "cool_backup",
"password": "hunter2",
"with_automatic_settings": True,
},
),
],
)
async def test_update_core_with_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
commands: list[dict[str, Any]],
default_mount: str | None,
expected_kwargs: dict[str, Any],
) -> None:
"""Test updating core update entity."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_ws_client(hass)
for command in commands:
await client.send_json_auto_id(command)
result = await client.receive_json()
assert result["success"]
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = default_mount
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await hass.services.async_call(
"update",
"install",
{"entity_id": "update.home_assistant_core_update", "backup": True},
blocking=True,
)
mock_create_backup.assert_called_once_with(**expected_kwargs)
supervisor_client.homeassistant.update.assert_called_once_with(
HomeAssistantUpdateOptions(version=None, backup=False)
)
supervisor_client.homeassistant.update.assert_called_once()
async def test_update_supervisor(
@ -325,6 +547,41 @@ async def test_update_addon_with_error(
)
async def test_update_addon_with_backup_and_error(
hass: HomeAssistant,
supervisor_client: AsyncMock,
) -> None:
"""Test updating addon update entity with error."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = None
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
side_effect=BackupManagerError,
),
pytest.raises(HomeAssistantError, match=r"^Error creating backup:"),
):
assert not await hass.services.async_call(
"update",
"install",
{"entity_id": "update.test_update", "backup": True},
blocking=True,
)
async def test_update_os_with_error(
hass: HomeAssistant, supervisor_client: AsyncMock
) -> None:
@ -406,6 +663,41 @@ async def test_update_core_with_error(
)
async def test_update_core_with_backup_and_error(
hass: HomeAssistant,
supervisor_client: AsyncMock,
) -> None:
"""Test updating core update entity with error."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = None
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
side_effect=BackupManagerError,
),
pytest.raises(HomeAssistantError, match=r"^Error creating backup:"),
):
assert not await hass.services.async_call(
"update",
"install",
{"entity_id": "update.home_assistant_core_update", "backup": True},
blocking=True,
)
async def test_release_notes_between_versions(
hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,

452
tests/components/hassio/test_websocket_api.py
View File

@ -1,9 +1,15 @@
"""Test websocket API."""
from unittest.mock import AsyncMock
import os
from typing import Any
from unittest.mock import AsyncMock, patch
from aiohasupervisor import SupervisorError
from aiohasupervisor.models import HomeAssistantUpdateOptions, StoreAddonUpdate
import pytest
from homeassistant.components.backup import BackupManagerError
from homeassistant.components.hassio import DOMAIN
from homeassistant.components.hassio.const import (
ATTR_DATA,
ATTR_ENDPOINT,
@ -15,14 +21,17 @@ from homeassistant.components.hassio.const import (
WS_TYPE_API,
WS_TYPE_SUBSCRIBE,
)
from homeassistant.const import __version__ as HAVERSION
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.setup import async_setup_component
from tests.common import MockUser, async_mock_signal
from tests.common import MockConfigEntry, MockUser, async_mock_signal
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import WebSocketGenerator
MOCK_ENVIRON = {"SUPERVISOR": "127.0.0.1", "SUPERVISOR_TOKEN": "abcdefgh"}
@pytest.fixture(autouse=True)
def mock_all(
@ -56,7 +65,7 @@ def mock_all(
)
aioclient_mock.get(
"http://127.0.0.1/core/info",
json={"result": "ok", "data": {"version_latest": "1.0.0"}},
json={"result": "ok", "data": {"version_latest": "1.0.0", "version": "1.0.0"}},
)
aioclient_mock.get(
"http://127.0.0.1/os/info",
@ -64,11 +73,42 @@ def mock_all(
)
aioclient_mock.get(
"http://127.0.0.1/supervisor/info",
json={"result": "ok", "data": {"version_latest": "1.0.0"}},
json={
"result": "ok",
"data": {
"version": "1.0.0",
"version_latest": "1.0.0",
"auto_update": True,
"addons": [
{
"name": "test",
"state": "started",
"slug": "test",
"installed": True,
"update_available": True,
"icon": False,
"version": "2.0.0",
"version_latest": "2.0.1",
"repository": "core",
"url": "https://github.com/home-assistant/addons/test",
},
],
},
},
)
aioclient_mock.get(
"http://127.0.0.1/ingress/panels", json={"result": "ok", "data": {"panels": {}}}
)
aioclient_mock.get(
"http://127.0.0.1/network/info",
json={
"result": "ok",
"data": {
"host_internet": True,
"supervisor_internet": True,
},
},
)
@pytest.mark.usefixtures("hassio_env")
@ -279,3 +319,407 @@ async def test_websocket_non_admin_user(
msg = await websocket_client.receive_json()
assert msg["error"]["message"] == "Unauthorized"
async def test_update_addon(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
update_addon: AsyncMock,
) -> None:
"""Test updating addon."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await client.send_json_auto_id(
{"type": "hassio/update/addon", "addon": "test", "backup": False}
)
result = await client.receive_json()
assert result["success"]
mock_create_backup.assert_not_called()
update_addon.assert_called_once_with("test", StoreAddonUpdate(backup=False))
@pytest.mark.parametrize(
("commands", "default_mount", "expected_kwargs"),
[
(
[],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": None,
},
),
(
[],
"my_nas",
{
"agent_ids": ["hassio.my_nas"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": None,
},
),
(
[
{
"type": "backup/config/update",
"create_backup": {
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"name": "cool_backup",
"password": "hunter2",
},
},
],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": ["test"],
"include_all_addons": False,
"include_database": False,
"include_folders": None,
"include_homeassistant": False,
"name": "test 2.0.0",
"password": "hunter2",
},
),
],
)
async def test_update_addon_with_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
update_addon: AsyncMock,
commands: list[dict[str, Any]],
default_mount: str | None,
expected_kwargs: dict[str, Any],
) -> None:
"""Test updating addon with backup."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_ws_client(hass)
for command in commands:
await client.send_json_auto_id(command)
result = await client.receive_json()
assert result["success"]
supervisor_client.mounts.info.return_value.default_backup_mount = default_mount
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await client.send_json_auto_id(
{"type": "hassio/update/addon", "addon": "test", "backup": True}
)
result = await client.receive_json()
assert result["success"]
mock_create_backup.assert_called_once_with(**expected_kwargs)
update_addon.assert_called_once_with("test", StoreAddonUpdate(backup=False))
async def test_update_core(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test updating core."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await client.send_json_auto_id({"type": "hassio/update/core", "backup": False})
result = await client.receive_json()
assert result["success"]
mock_create_backup.assert_not_called()
supervisor_client.homeassistant.update.assert_called_once_with(
HomeAssistantUpdateOptions(version=None, backup=False)
)
@pytest.mark.parametrize(
("commands", "default_mount", "expected_kwargs"),
[
(
[],
None,
{
"agent_ids": ["hassio.local"],
"include_addons": None,
"include_all_addons": False,
"include_database": True,
"include_folders": None,
"include_homeassistant": True,
"name": f"Home Assistant Core {HAVERSION}",
"password": None,
},
),
(
[],
"my_nas",
{
"agent_ids": ["hassio.my_nas"],
"include_addons": None,
"include_all_addons": False,
"include_database": True,
"include_folders": None,
"include_homeassistant": True,
"name": f"Home Assistant Core {HAVERSION}",
"password": None,
},
),
(
[
{
"type": "backup/config/update",
"create_backup": {
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"name": "cool_backup",
"password": "hunter2",
},
},
],
None,
{
"agent_ids": ["test-agent"],
"include_addons": ["my-addon"],
"include_all_addons": True,
"include_database": False,
"include_folders": ["share"],
"include_homeassistant": True,
"name": "cool_backup",
"password": "hunter2",
"with_automatic_settings": True,
},
),
],
)
async def test_update_core_with_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
commands: list[dict[str, Any]],
default_mount: str | None,
expected_kwargs: dict[str, Any],
) -> None:
"""Test updating core with backup."""
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
client = await hass_ws_client(hass)
for command in commands:
await client.send_json_auto_id(command)
result = await client.receive_json()
assert result["success"]
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = default_mount
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
) as mock_create_backup:
await client.send_json_auto_id({"type": "hassio/update/core", "backup": True})
result = await client.receive_json()
assert result["success"]
mock_create_backup.assert_called_once_with(**expected_kwargs)
supervisor_client.homeassistant.update.assert_called_once_with(
HomeAssistantUpdateOptions(version=None, backup=False)
)
async def test_update_addon_with_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
update_addon: AsyncMock,
) -> None:
"""Test updating addon with error."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
assert await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
await hass.async_block_till_done()
update_addon.side_effect = SupervisorError
await client.send_json_auto_id(
{"type": "hassio/update/addon", "addon": "test", "backup": False}
)
result = await client.receive_json()
assert not result["success"]
assert result["error"] == {
"code": "home_assistant_error",
"message": "Error updating test: ",
}
async def test_update_addon_with_backup_and_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test updating addon with backup and error."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = None
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
side_effect=BackupManagerError,
),
):
await client.send_json_auto_id(
{"type": "hassio/update/addon", "addon": "test", "backup": True}
)
result = await client.receive_json()
assert not result["success"]
assert result["error"] == {
"code": "home_assistant_error",
"message": "Error creating backup: ",
}
async def test_update_core_with_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test updating core with error."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
assert await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
await hass.async_block_till_done()
supervisor_client.homeassistant.update.side_effect = SupervisorError
await client.send_json_auto_id({"type": "hassio/update/core", "backup": False})
result = await client.receive_json()
assert not result["success"]
assert result["error"] == {
"code": "home_assistant_error",
"message": "Error updating Home Assistant Core: ",
}
async def test_update_core_with_backup_and_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test updating core with backup and error."""
client = await hass_ws_client(hass)
config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN)
config_entry.add_to_hass(hass)
with patch.dict(os.environ, MOCK_ENVIRON):
result = await async_setup_component(
hass,
"hassio",
{"http": {"server_port": 9999, "server_host": "127.0.0.1"}, "hassio": {}},
)
assert result
assert await async_setup_component(hass, "backup", {})
await hass.async_block_till_done()
supervisor_client.homeassistant.update.return_value = None
supervisor_client.mounts.info.return_value.default_backup_mount = None
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
side_effect=BackupManagerError,
),
):
await client.send_json_auto_id(
{"type": "hassio/update/addon", "addon": "test", "backup": True}
)
result = await client.receive_json()
assert not result["success"]
assert result["error"] == {
"code": "home_assistant_error",
"message": "Error creating backup: ",
}