hass-core/tests/components/backup/test_websocket.py
Joakim Sørensen 31dcc25ba5
Add handler to restore a backup file with the backup integration (#128365)
* Early pushout of restore handling for core/container

* Adjust after rebase

* Move logging definition, we should only do this if we go ahead with the restore

* First round

* More paths

* Add async_restore_backup to base class

* Block restore of new backup files

* manager tests

* Add websocket test

* Add testing to main

* Add coverage for missing backup file

* Catch FileNotFoundError instead

* Patch Path.read_text instead

* Remove HA_RESTORE from keep

* Use secure paths

* Fix restart test

* extend coverage

* Mock argv

* Adjustments
2024-11-01 16:25:22 +01:00

291 lines
8.3 KiB
Python

"""Tests for the Backup integration."""
from unittest.mock import patch
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.backup.manager import Backup
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .common import TEST_BACKUP, setup_backup_integration
from tests.typing import WebSocketGenerator
@pytest.fixture
def sync_access_token_proxy(
access_token_fixture_name: str,
request: pytest.FixtureRequest,
) -> str:
"""Non-async proxy for the *_access_token fixture.
Workaround for https://github.com/pytest-dev/pytest-asyncio/issues/112
"""
return request.getfixturevalue(access_token_fixture_name)
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
) -> None:
"""Test getting backup info."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backups",
return_value={TEST_BACKUP.slug: TEST_BACKUP},
):
await client.send_json_auto_id({"type": "backup/info"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"backup_content",
[
pytest.param(TEST_BACKUP, id="with_backup_content"),
pytest.param(None, id="without_backup_content"),
],
)
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_details(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
backup_content: Backup | None,
) -> None:
"""Test getting backup info."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
return_value=backup_content,
):
await client.send_json_auto_id({"type": "backup/details", "slug": "abc123"})
assert await client.receive_json() == snapshot
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_remove(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
) -> None:
"""Test removing a backup file."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_remove_backup",
):
await client.send_json_auto_id({"type": "backup/remove", "slug": "abc123"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_generate(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
) -> None:
"""Test generating a backup."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_create_backup",
return_value=TEST_BACKUP,
):
await client.send_json_auto_id({"type": "backup/generate"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_restore(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
) -> None:
"""Test calling the restore command."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
):
await client.send_json_auto_id({"type": "backup/restore", "slug": "abc123"})
assert await client.receive_json() == snapshot
@pytest.mark.parametrize(
"access_token_fixture_name",
["hass_access_token", "hass_supervisor_access_token"],
)
@pytest.mark.parametrize(
("with_hassio"),
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_backup_end(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
sync_access_token_proxy: str,
*,
access_token_fixture_name: str,
with_hassio: bool,
) -> None:
"""Test handling of post backup actions from a WS command."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass, sync_access_token_proxy)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_post_backup_actions",
):
await client.send_json_auto_id({"type": "backup/end"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"access_token_fixture_name",
["hass_access_token", "hass_supervisor_access_token"],
)
@pytest.mark.parametrize(
("with_hassio"),
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_backup_start(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
sync_access_token_proxy: str,
*,
access_token_fixture_name: str,
with_hassio: bool,
) -> None:
"""Test handling of pre backup actions from a WS command."""
await setup_backup_integration(hass, with_hassio=with_hassio)
client = await hass_ws_client(hass, sync_access_token_proxy)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_pre_backup_actions",
):
await client.send_json_auto_id({"type": "backup/start"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"exception",
[
TimeoutError(),
HomeAssistantError("Boom"),
Exception("Boom"),
],
)
async def test_backup_end_excepion(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
hass_supervisor_access_token: str,
exception: Exception,
) -> None:
"""Test exception handling while running post backup actions from a WS command."""
await setup_backup_integration(hass, with_hassio=True)
client = await hass_ws_client(hass, hass_supervisor_access_token)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_post_backup_actions",
side_effect=exception,
):
await client.send_json_auto_id({"type": "backup/end"})
assert snapshot == await client.receive_json()
@pytest.mark.parametrize(
"exception",
[
TimeoutError(),
HomeAssistantError("Boom"),
Exception("Boom"),
],
)
async def test_backup_start_excepion(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
hass_supervisor_access_token: str,
exception: Exception,
) -> None:
"""Test exception handling while running pre backup actions from a WS command."""
await setup_backup_integration(hass, with_hassio=True)
client = await hass_ws_client(hass, hass_supervisor_access_token)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_pre_backup_actions",
side_effect=exception,
):
await client.send_json_auto_id({"type": "backup/start"})
assert snapshot == await client.receive_json()