"""Tests for the Backup integration.""" from collections.abc import Generator from typing import Any from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch from freezegun.api import FrozenDateTimeFactory import pytest from pytest_unordered import unordered from syrupy import SnapshotAssertion from homeassistant.components.backup import ( AgentBackup, BackupAgentError, BackupAgentPlatformProtocol, BackupNotFound, BackupReaderWriterError, Folder, store, ) from homeassistant.components.backup.agent import BackupAgentUnreachableError from homeassistant.components.backup.const import DATA_MANAGER, DOMAIN from homeassistant.components.backup.manager import ( AgentBackupStatus, CreateBackupEvent, CreateBackupState, ManagerBackup, NewBackup, ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.setup import async_setup_component from .common import ( LOCAL_AGENT_ID, TEST_BACKUP_ABC123, TEST_BACKUP_DEF456, BackupAgentTest, setup_backup_integration, setup_backup_platform, ) from tests.common import async_fire_time_changed, async_mock_service from tests.typing import WebSocketGenerator BACKUP_CALL = call( agent_ids=["test.test-agent"], backup_name="test-name", extra_metadata={"instance_id": ANY, "with_automatic_settings": True}, include_addons=["test-addon"], include_all_addons=False, include_database=True, include_folders=["media"], include_homeassistant=True, password="test-password", on_progress=ANY, ) DEFAULT_STORAGE_DATA: dict[str, Any] = { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": [], "include_addons": None, "include_all_addons": False, "include_database": True, "include_folders": None, "name": None, "password": None, }, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "retention": { "copies": None, "days": None, }, "schedule": {"days": [], "recurrence": "never", "state": "never", "time": None}, }, } DAILY = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] @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.fixture(autouse=True) def mock_delay_save() -> Generator[None]: """Mock the delay save constant.""" with patch("homeassistant.components.backup.store.STORE_DELAY_SAVE", 0): yield @pytest.fixture(name="delete_backup") def mock_delete_backup() -> Generator[AsyncMock]: """Mock manager delete backup.""" with patch( "homeassistant.components.backup.BackupManager.async_delete_backup" ) as mock_delete_backup: yield mock_delete_backup @pytest.fixture(name="get_backups") def mock_get_backups() -> Generator[AsyncMock]: """Mock manager get backups.""" with patch( "homeassistant.components.backup.BackupManager.async_get_backups" ) as mock_get_backups: yield mock_get_backups @pytest.mark.parametrize( ("remote_agents", "remote_backups"), [ ([], {}), (["remote"], {}), (["remote"], {"test.remote": [TEST_BACKUP_ABC123]}), (["remote"], {"test.remote": [TEST_BACKUP_DEF456]}), ], ) async def test_info( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, remote_agents: list[str], remote_backups: dict[str, list[AgentBackup]], snapshot: SnapshotAssertion, ) -> None: """Test getting backup info.""" await setup_backup_integration( hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]} | remote_backups, remote_agents=remote_agents, ) client = await hass_ws_client(hass) await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/info"}) assert await client.receive_json() == snapshot @pytest.mark.parametrize( "side_effect", [Exception("Oops"), HomeAssistantError("Boom!"), BackupAgentUnreachableError], ) async def test_info_with_errors( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, side_effect: Exception, snapshot: SnapshotAssertion, ) -> None: """Test getting backup info with one unavailable agent.""" await setup_backup_integration( hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]} ) hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test") client = await hass_ws_client(hass) await hass.async_block_till_done() with patch.object(BackupAgentTest, "async_list_backups", side_effect=side_effect): await client.send_json_auto_id({"type": "backup/info"}) assert await client.receive_json() == snapshot @pytest.mark.parametrize( ("remote_agents", "backups"), [ ([], {}), (["remote"], {LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]}), (["remote"], {"test.remote": [TEST_BACKUP_ABC123]}), (["remote"], {"test.remote": [TEST_BACKUP_DEF456]}), ( ["remote"], { LOCAL_AGENT_ID: [TEST_BACKUP_ABC123], "test.remote": [TEST_BACKUP_ABC123], }, ), ], ) async def test_details( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, remote_agents: list[str], backups: dict[str, list[AgentBackup]], snapshot: SnapshotAssertion, ) -> None: """Test getting backup info.""" await setup_backup_integration( hass, with_hassio=False, backups=backups, remote_agents=remote_agents ) client = await hass_ws_client(hass) await hass.async_block_till_done() with patch("pathlib.Path.exists", return_value=True): await client.send_json_auto_id( {"type": "backup/details", "backup_id": "abc123"} ) assert await client.receive_json() == snapshot @pytest.mark.parametrize( "side_effect", [Exception("Oops"), HomeAssistantError("Boom!"), BackupAgentUnreachableError], ) async def test_details_with_errors( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, side_effect: Exception, snapshot: SnapshotAssertion, ) -> None: """Test getting backup info with one unavailable agent.""" await setup_backup_integration( hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]} ) hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test") client = await hass_ws_client(hass) await hass.async_block_till_done() with ( patch("pathlib.Path.exists", return_value=True), patch.object(BackupAgentTest, "async_get_backup", side_effect=side_effect), ): await client.send_json_auto_id( {"type": "backup/details", "backup_id": "abc123"} ) assert await client.receive_json() == snapshot @pytest.mark.parametrize( ("remote_agents", "backups"), [ ([], {}), (["remote"], {LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]}), (["remote"], {"test.remote": [TEST_BACKUP_ABC123]}), (["remote"], {"test.remote": [TEST_BACKUP_DEF456]}), ( ["remote"], { LOCAL_AGENT_ID: [TEST_BACKUP_ABC123], "test.remote": [TEST_BACKUP_ABC123], }, ), ], ) async def test_delete( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, remote_agents: list[str], backups: dict[str, list[AgentBackup]], snapshot: SnapshotAssertion, ) -> None: """Test deleting a backup file.""" await setup_backup_integration( hass, with_hassio=False, backups=backups, remote_agents=remote_agents ) client = await hass_ws_client(hass) await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/info"}) assert await client.receive_json() == snapshot await client.send_json_auto_id({"type": "backup/delete", "backup_id": "abc123"}) assert await client.receive_json() == snapshot await client.send_json_auto_id({"type": "backup/info"}) assert await client.receive_json() == snapshot @pytest.mark.parametrize( "storage_data", [ DEFAULT_STORAGE_DATA, DEFAULT_STORAGE_DATA | { "backups": [ { "backup_id": "abc123", "failed_agent_ids": ["test.remote"], } ] }, ], ) @pytest.mark.parametrize( "side_effect", [None, HomeAssistantError("Boom!"), BackupAgentUnreachableError] ) async def test_delete_with_errors( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_storage: dict[str, Any], side_effect: Exception, storage_data: dict[str, Any] | None, snapshot: SnapshotAssertion, ) -> None: """Test deleting a backup with one unavailable agent.""" hass_storage[DOMAIN] = { "data": storage_data, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, } await setup_backup_integration( hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]} ) hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test") client = await hass_ws_client(hass) await hass.async_block_till_done() with patch.object(BackupAgentTest, "async_delete_backup", side_effect=side_effect): await client.send_json_auto_id({"type": "backup/delete", "backup_id": "abc123"}) assert await client.receive_json() == snapshot await client.send_json_auto_id({"type": "backup/info"}) assert await client.receive_json() == snapshot async def test_agent_delete_backup( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, ) -> None: """Test deleting a backup file with a mock agent.""" await setup_backup_integration(hass) hass.data[DATA_MANAGER].backup_agents = {"domain.test": BackupAgentTest("test")} client = await hass_ws_client(hass) await hass.async_block_till_done() with patch.object(BackupAgentTest, "async_delete_backup") as delete_mock: await client.send_json_auto_id( { "type": "backup/delete", "backup_id": "abc123", } ) assert await client.receive_json() == snapshot assert delete_mock.call_args == call("abc123") @pytest.mark.parametrize( "data", [ None, {}, {"password": "abc123"}, ], ) @pytest.mark.usefixtures("mock_backup_generation") async def test_generate( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, data: dict[str, Any] | None, freezer: FrozenDateTimeFactory, snapshot: SnapshotAssertion, ) -> None: """Test generating a backup.""" await setup_backup_integration(hass, with_hassio=False) client = await hass_ws_client(hass) freezer.move_to("2024-11-13 12:01:00+01:00") await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/subscribe_events"}) assert await client.receive_json() == snapshot await client.send_json_auto_id( {"type": "backup/generate", **{"agent_ids": ["backup.local"]} | (data or {})} ) for _ in range(6): assert await client.receive_json() == snapshot @pytest.mark.parametrize( ("parameters", "expected_error"), [ ( {"include_homeassistant": False}, "Home Assistant must be included in backup", ), ( {"include_addons": ["blah"]}, "Addons and folders are not supported by core backup", ), ( {"include_all_addons": True}, "Addons and folders are not supported by core backup", ), ( {"include_folders": ["ssl"]}, "Addons and folders are not supported by core backup", ), ], ) async def test_generate_wrong_parameters( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, parameters: dict[str, Any], expected_error: str, ) -> None: """Test generating a backup.""" await setup_backup_integration(hass, with_hassio=False) client = await hass_ws_client(hass) default_parameters = {"type": "backup/generate", "agent_ids": ["backup.local"]} await client.send_json_auto_id(default_parameters | parameters) response = await client.receive_json() assert not response["success"] assert response["error"] == { "code": "home_assistant_error", "message": expected_error, } @pytest.mark.usefixtures("mock_backup_generation") @pytest.mark.parametrize( ("params", "expected_extra_call_params"), [ ({"agent_ids": ["backup.local"]}, {"agent_ids": ["backup.local"]}), ( { "agent_ids": ["backup.local"], "include_database": False, "name": "abc123", }, { "agent_ids": ["backup.local"], "include_addons": None, "include_database": False, "include_folders": None, "name": "abc123", }, ), ], ) async def test_generate_calls_create( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, snapshot: SnapshotAssertion, params: dict[str, Any], expected_extra_call_params: dict[str, Any], ) -> None: """Test translation of WS parameter to backup/generate to async_initiate_backup.""" await setup_backup_integration(hass, with_hassio=False) client = await hass_ws_client(hass) freezer.move_to("2024-11-13 12:01:00+01:00") await hass.async_block_till_done() with patch( "homeassistant.components.backup.manager.BackupManager.async_initiate_backup", return_value=NewBackup(backup_job_id="abc123"), ) as generate_backup: await client.send_json_auto_id({"type": "backup/generate"} | params) result = await client.receive_json() assert result["success"] assert result["result"] == {"backup_job_id": "abc123"} generate_backup.assert_called_once_with( **{ "include_all_addons": False, "include_homeassistant": True, "include_addons": None, "include_database": True, "include_folders": None, "name": None, "password": None, } | expected_extra_call_params ) @pytest.mark.parametrize( ( "create_backup_settings", "expected_call_params", "side_effect", "last_completed_automatic_backup", ), [ ( { "agent_ids": ["test.remote"], "include_addons": None, "include_all_addons": False, "include_database": True, "include_folders": None, "name": None, "password": None, }, { "agent_ids": ["test.remote"], "backup_name": ANY, "extra_metadata": { "instance_id": ANY, "with_automatic_settings": True, }, "include_addons": None, "include_all_addons": False, "include_database": True, "include_folders": None, "include_homeassistant": True, "on_progress": ANY, "password": None, }, None, "2024-11-13T12:01:01+01:00", ), ( { "agent_ids": ["test.remote"], "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, { "agent_ids": ["test.remote"], "backup_name": "test-name", "extra_metadata": { "instance_id": ANY, "with_automatic_settings": True, }, "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": [Folder.MEDIA], "include_homeassistant": True, "on_progress": ANY, "password": "test-password", }, None, "2024-11-13T12:01:01+01:00", ), ( { "agent_ids": ["test.remote"], "include_addons": None, "include_all_addons": False, "include_database": True, "include_folders": None, "name": None, "password": None, }, { "agent_ids": ["test.remote"], "backup_name": ANY, "extra_metadata": { "instance_id": ANY, "with_automatic_settings": True, }, "include_addons": None, "include_all_addons": False, "include_database": True, "include_folders": None, "include_homeassistant": True, "on_progress": ANY, "password": None, }, BackupAgentError("Boom!"), None, ), ], ) async def test_generate_with_default_settings_calls_create( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_storage: dict[str, Any], freezer: FrozenDateTimeFactory, create_backup: AsyncMock, create_backup_settings: dict[str, Any], expected_call_params: dict[str, Any], side_effect: Exception | None, last_completed_automatic_backup: str, ) -> None: """Test backup/generate_with_automatic_settings calls async_initiate_backup.""" created_backup: MagicMock = create_backup.return_value[1].result().backup created_backup.protected = create_backup_settings["password"] is not None client = await hass_ws_client(hass) await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-13T12:01:00+01:00") remote_agent = BackupAgentTest("remote", backups=[]) await setup_backup_platform( hass, domain="test", platform=Mock( async_get_backup_agents=AsyncMock(return_value=[remote_agent]), spec_set=BackupAgentPlatformProtocol, ), ) await async_setup_component(hass, DOMAIN, {}) await hass.async_block_till_done() await client.send_json_auto_id( {"type": "backup/config/update", "create_backup": create_backup_settings} ) result = await client.receive_json() assert result["success"] freezer.tick() async_fire_time_changed(hass) await hass.async_block_till_done() assert ( hass_storage[DOMAIN]["data"]["config"]["create_backup"] == create_backup_settings ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_attempted_automatic_backup"] is None ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_completed_automatic_backup"] is None ) with patch.object(remote_agent, "async_upload_backup", side_effect=side_effect): await client.send_json_auto_id( {"type": "backup/generate_with_automatic_settings"} ) result = await client.receive_json() assert result["success"] assert result["result"] == {"backup_job_id": "abc123"} await hass.async_block_till_done() create_backup.assert_called_once_with(**expected_call_params) freezer.tick() async_fire_time_changed(hass) await hass.async_block_till_done() assert ( hass_storage[DOMAIN]["data"]["config"]["last_attempted_automatic_backup"] == "2024-11-13T12:01:01+01:00" ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_completed_automatic_backup"] == last_completed_automatic_backup ) @pytest.mark.parametrize( "backups", [ {}, {LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]}, ], ) async def test_restore_local_agent( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, backups: dict[str, list[AgentBackup]], snapshot: SnapshotAssertion, ) -> None: """Test calling the restore command.""" await setup_backup_integration(hass, with_hassio=False, backups=backups) restart_calls = async_mock_service(hass, "homeassistant", "restart") client = await hass_ws_client(hass) await hass.async_block_till_done() with ( patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.write_text"), patch("homeassistant.components.backup.manager.validate_password"), ): await client.send_json_auto_id( { "type": "backup/restore", "backup_id": "abc123", "agent_id": "backup.local", } ) assert await client.receive_json() == snapshot assert len(restart_calls) == snapshot @pytest.mark.parametrize( ("remote_agents", "backups"), [ (["remote"], {}), (["remote"], {"test.remote": [TEST_BACKUP_ABC123]}), ], ) async def test_restore_remote_agent( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, remote_agents: list[str], backups: dict[str, list[AgentBackup]], snapshot: SnapshotAssertion, ) -> None: """Test calling the restore command.""" await setup_backup_integration( hass, with_hassio=False, backups=backups, remote_agents=remote_agents ) restart_calls = async_mock_service(hass, "homeassistant", "restart") client = await hass_ws_client(hass) await hass.async_block_till_done() with ( patch("pathlib.Path.write_text"), patch("pathlib.Path.open"), patch("homeassistant.components.backup.manager.validate_password"), ): await client.send_json_auto_id( { "type": "backup/restore", "backup_id": "abc123", "agent_id": "test.remote", } ) assert await client.receive_json() == snapshot assert len(restart_calls) == snapshot async def test_restore_wrong_password( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, ) -> None: """Test calling the restore command.""" await setup_backup_integration( hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BACKUP_ABC123]} ) restart_calls = async_mock_service(hass, "homeassistant", "restart") client = await hass_ws_client(hass) await hass.async_block_till_done() with ( patch("pathlib.Path.exists", return_value=True), patch("pathlib.Path.write_text"), patch( "homeassistant.components.backup.manager.validate_password", return_value=False, ), ): await client.send_json_auto_id( { "type": "backup/restore", "backup_id": "abc123", "agent_id": "backup.local", } ) assert await client.receive_json() == snapshot assert len(restart_calls) == 0 @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"), ], ) @pytest.mark.usefixtures("supervisor_client") 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 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"), ], ) @pytest.mark.usefixtures("supervisor_client") 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 await client.receive_json() == snapshot @pytest.mark.parametrize( "exception", [ TimeoutError(), HomeAssistantError("Boom"), Exception("Boom"), ], ) @pytest.mark.usefixtures("supervisor_client") async def test_backup_end_exception( 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 await client.receive_json() == snapshot @pytest.mark.parametrize( "exception", [ TimeoutError(), HomeAssistantError("Boom"), Exception("Boom"), ], ) @pytest.mark.usefixtures("supervisor_client") async def test_backup_start_exception( 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 await client.receive_json() == snapshot async def test_agents_info( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, ) -> None: """Test getting backup agents info.""" await setup_backup_integration(hass, with_hassio=False) hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test") client = await hass_ws_client(hass) await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/agents/info"}) assert await client.receive_json() == snapshot @pytest.mark.usefixtures("create_backup", "delete_backup", "get_backups") @pytest.mark.parametrize( "storage_data", [ {}, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": ["test-addon"], "include_all_addons": True, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "retention": {"copies": 3, "days": 7}, "last_attempted_automatic_backup": "2024-10-26T04:45:00+01:00", "last_completed_automatic_backup": "2024-10-26T04:45:00+01:00", "schedule": { "days": DAILY, "recurrence": "custom_days", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": 3, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": [], "recurrence": "never", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": None, "days": 7}, "last_attempted_automatic_backup": "2024-10-27T04:45:00+01:00", "last_completed_automatic_backup": "2024-10-26T04:45:00+01:00", "schedule": { "days": [], "recurrence": "never", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": ["mon"], "recurrence": "custom_days", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": [], "recurrence": "never", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": ["mon", "sun"], "recurrence": "custom_days", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, { "backup": { "data": { "backups": [], "config": { "agents": { "test-agent1": {"protected": True}, "test-agent2": {"protected": False}, }, "create_backup": { "agent_ids": ["test-agent"], "include_addons": None, "include_all_addons": False, "include_database": False, "include_folders": None, "name": None, "password": None, }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": ["mon", "sun"], "recurrence": "custom_days", "state": "never", "time": None, }, }, }, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, }, }, ], ) @patch("homeassistant.components.backup.config.random.randint", Mock(return_value=600)) async def test_config_info( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, snapshot: SnapshotAssertion, hass_storage: dict[str, Any], storage_data: dict[str, Any] | None, ) -> None: """Test getting backup config info.""" client = await hass_ws_client(hass) await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-13T12:01:00+01:00") hass_storage.update(storage_data) await setup_backup_integration(hass) await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/config/info"}) assert await client.receive_json() == snapshot @pytest.mark.usefixtures("create_backup", "delete_backup", "get_backups") @pytest.mark.parametrize( "commands", [ [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 7}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "schedule": {"recurrence": "daily", "time": "06:00"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "schedule": {"days": ["mon"], "recurrence": "custom_days"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "schedule": {"recurrence": "never"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "schedule": {"days": ["mon", "sun"], "recurrence": "custom_days"}, } ], [ { "type": "backup/config/update", "create_backup": { "agent_ids": ["test-agent"], "include_addons": ["test-addon"], "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": 3, "days": 7}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": None}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 7}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": 3}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"days": 7}, "schedule": {"recurrence": "daily"}, } ], [ { "type": "backup/config/update", "agents": { "test-agent1": {"protected": True}, "test-agent2": {"protected": False}, }, } ], [ # Test we can update AgentConfig { "type": "backup/config/update", "agents": { "test-agent1": {"protected": True}, "test-agent2": {"protected": False}, }, }, { "type": "backup/config/update", "agents": { "test-agent1": {"protected": False}, "test-agent2": {"protected": True}, }, }, ], ], ) @patch("homeassistant.components.backup.config.random.randint", Mock(return_value=600)) async def test_config_update( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, snapshot: SnapshotAssertion, commands: dict[str, Any], hass_storage: dict[str, Any], ) -> None: """Test updating the backup config.""" client = await hass_ws_client(hass) await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-13T12:01:00+01:00") await setup_backup_integration(hass) await hass.async_block_till_done() await client.send_json_auto_id({"type": "backup/config/info"}) assert await client.receive_json() == snapshot for command in commands: await client.send_json_auto_id(command) result = await client.receive_json() assert result["success"] await client.send_json_auto_id({"type": "backup/config/info"}) assert await client.receive_json() == snapshot await hass.async_block_till_done() # Trigger store write freezer.tick(60) async_fire_time_changed(hass) await hass.async_block_till_done() assert hass_storage[DOMAIN] == snapshot @pytest.mark.usefixtures("create_backup", "delete_backup", "get_backups") @pytest.mark.parametrize( "command", [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": "blah", }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": "never", }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": {"state": "someday"}, }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": {"time": "early"}, }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": {"days": "mon"}, }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "recurrence": {"days": ["fun"]}, }, { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent", "test-agent"]}, }, { "type": "backup/config/update", "create_backup": {"include_addons": ["my-addon", "my-addon"]}, }, { "type": "backup/config/update", "create_backup": {"include_folders": ["media", "media"]}, }, { "type": "backup/config/update", "agents": {"test-agent1": {"favorite": True}}, }, ], ) async def test_config_update_errors( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, command: dict[str, Any], ) -> None: """Test errors when updating the backup config.""" await setup_backup_integration(hass) await hass.async_block_till_done() client = await hass_ws_client(hass) await client.send_json_auto_id({"type": "backup/config/info"}) assert await client.receive_json() == snapshot await client.send_json_auto_id(command) result = await client.receive_json() assert not result["success"] await client.send_json_auto_id({"type": "backup/config/info"}) assert await client.receive_json() == snapshot await hass.async_block_till_done() @pytest.mark.parametrize( ( "commands", "last_completed_automatic_backup", "time_1", "time_2", "attempted_backup_time", "completed_backup_time", "scheduled_backup_time", "additional_backup", "backup_calls_1", "backup_calls_2", "call_args", "create_backup_side_effect", ), [ ( # No config update [], "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( # Unchanged schedule [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "daily"}, } ], "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"days": ["mon"], "recurrence": "custom_days"}, } ], "2024-11-11T04:45:00+01:00", "2024-11-18T04:55:00+01:00", "2024-11-25T04:55:00+01:00", "2024-11-18T04:55:00+01:00", "2024-11-18T04:55:00+01:00", "2024-11-18T04:55:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": { "days": ["mon"], "recurrence": "custom_days", "time": "03:45", }, } ], "2024-11-11T03:45:00+01:00", "2024-11-18T03:45:00+01:00", "2024-11-25T03:45:00+01:00", "2024-11-18T03:45:00+01:00", "2024-11-18T03:45:00+01:00", "2024-11-18T03:45:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "daily", "time": "03:45"}, } ], "2024-11-11T03:45:00+01:00", "2024-11-12T03:45:00+01:00", "2024-11-13T03:45:00+01:00", "2024-11-12T03:45:00+01:00", "2024-11-12T03:45:00+01:00", "2024-11-12T03:45:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"days": ["wed", "fri"], "recurrence": "custom_days"}, } ], "2024-11-11T04:45:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-15T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-13T04:55:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "never"}, } ], "2024-11-11T04:45:00+01:00", "2034-11-11T12:00:00+01:00", # ten years later and still no backups "2034-11-11T13:00:00+01:00", "2024-11-11T04:45:00+01:00", "2024-11-11T04:45:00+01:00", None, False, 0, 0, None, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"days": [], "recurrence": "custom_days"}, } ], "2024-11-11T04:45:00+01:00", "2034-11-11T12:00:00+01:00", # ten years later and still no backups "2034-11-11T13:00:00+01:00", "2024-11-11T04:45:00+01:00", "2024-11-11T04:45:00+01:00", None, False, 0, 0, None, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "daily"}, } ], "2024-10-26T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-12T04:55:00+01:00", False, 1, 2, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"days": ["mon"], "recurrence": "custom_days"}, } ], "2024-10-26T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", # missed event uses daily schedule once "2024-11-12T04:55:00+01:00", # missed event uses daily schedule once "2024-11-12T04:55:00+01:00", True, 1, 1, BACKUP_CALL, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "never"}, } ], "2024-10-26T04:45:00+01:00", "2034-11-11T12:00:00+01:00", # ten years later and still no backups "2034-11-12T12:00:00+01:00", "2024-10-26T04:45:00+01:00", "2024-10-26T04:45:00+01:00", None, False, 0, 0, None, None, ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "daily"}, } ], "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", # attempted to create backup but failed "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", False, 1, 2, BACKUP_CALL, [BackupReaderWriterError("Boom"), None], ), ( [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "schedule": {"recurrence": "daily"}, } ], "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", "2024-11-13T04:55:00+01:00", "2024-11-12T04:55:00+01:00", # attempted to create backup but failed "2024-11-11T04:45:00+01:00", "2024-11-12T04:55:00+01:00", False, 1, 2, BACKUP_CALL, [Exception("Boom"), None], # unknown error ), ], ) @patch("homeassistant.components.backup.config.random.randint", Mock(return_value=600)) async def test_config_schedule_logic( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, hass_storage: dict[str, Any], create_backup: AsyncMock, commands: list[dict[str, Any]], last_completed_automatic_backup: str, time_1: str, time_2: str, attempted_backup_time: str, completed_backup_time: str, scheduled_backup_time: str, additional_backup: bool, backup_calls_1: int, backup_calls_2: int, call_args: Any, create_backup_side_effect: list[Exception | None] | None, ) -> None: """Test config schedule logic.""" created_backup: MagicMock = create_backup.return_value[1].result().backup created_backup.protected = True client = await hass_ws_client(hass) storage_data = { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test.test-agent"], "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": last_completed_automatic_backup, "last_completed_automatic_backup": last_completed_automatic_backup, "schedule": { "days": [], "recurrence": "daily", "state": "never", "time": None, }, }, } hass_storage[DOMAIN] = { "data": storage_data, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, } create_backup.side_effect = create_backup_side_effect await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-11 12:00:00+01:00") await setup_backup_integration(hass, remote_agents=["test-agent"]) await hass.async_block_till_done() for command in commands: await client.send_json_auto_id(command) result = await client.receive_json() assert result["success"] await client.send_json_auto_id({"type": "backup/info"}) result = await client.receive_json() assert result["result"]["next_automatic_backup"] == scheduled_backup_time assert result["result"]["next_automatic_backup_additional"] == additional_backup freezer.move_to(time_1) async_fire_time_changed(hass) await hass.async_block_till_done() assert create_backup.call_count == backup_calls_1 assert create_backup.call_args == call_args async_fire_time_changed(hass, fire_all=True) # flush out storage save await hass.async_block_till_done() assert ( hass_storage[DOMAIN]["data"]["config"]["last_attempted_automatic_backup"] == attempted_backup_time ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_completed_automatic_backup"] == completed_backup_time ) freezer.move_to(time_2) async_fire_time_changed(hass) await hass.async_block_till_done() assert create_backup.call_count == backup_calls_2 assert create_backup.call_args == call_args @pytest.mark.parametrize( ( "command", "backups", "get_backups_agent_errors", "delete_backup_agent_errors", "last_backup_time", "next_time", "backup_time", "backup_calls", "get_backups_calls", "delete_calls", "delete_args_list", ), [ ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": None, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, # we get backups even if backup retention copies is None 0, [], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 0, [], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 0, [], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 1, [ call( "backup-1", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ) ], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 1, [ call( "backup-1", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ) ], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 2, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 2, [ call("backup-1", agent_ids=["test.test-agent"]), call("backup-2", agent_ids=["test.test-agent"]), ], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 2, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {"test-agent": BackupAgentError("Boom!")}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 2, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {"test-agent": BackupAgentError("Boom!")}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 0, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 3, [ call( "backup-1", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ), call( "backup-2", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ), call( "backup-3", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ), ], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 0, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={ "test.test-agent": MagicMock(spec=AgentBackupStatus), "test.test-agent2": MagicMock(spec=AgentBackupStatus), }, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 3, [ call( "backup-1", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ), call( "backup-2", agent_ids=unordered(["test.test-agent", "test.test-agent2"]), ), call("backup-3", agent_ids=["test.test-agent"]), ], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 0, "days": None}, "schedule": {"recurrence": "daily"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-12T04:45:00+01:00", "2024-11-12T04:45:00+01:00", 1, 1, 0, [], ), ], ) @patch("homeassistant.components.backup.config.BACKUP_START_TIME_JITTER", 0) async def test_config_retention_copies_logic( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, hass_storage: dict[str, Any], create_backup: AsyncMock, delete_backup: AsyncMock, get_backups: AsyncMock, command: dict[str, Any], backups: dict[str, Any], get_backups_agent_errors: dict[str, Exception], delete_backup_agent_errors: dict[str, Exception], last_backup_time: str, next_time: str, backup_time: str, backup_calls: int, get_backups_calls: int, delete_calls: int, delete_args_list: Any, ) -> None: """Test config backup retention copies logic.""" created_backup: MagicMock = create_backup.return_value[1].result().backup created_backup.protected = True client = await hass_ws_client(hass) storage_data = { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": last_backup_time, "schedule": { "days": [], "recurrence": "daily", "state": "never", "time": None, }, }, } hass_storage[DOMAIN] = { "data": storage_data, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, } get_backups.return_value = (backups, get_backups_agent_errors) delete_backup.return_value = delete_backup_agent_errors await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-11 12:00:00+01:00") await setup_backup_integration(hass, remote_agents=["test-agent"]) await hass.async_block_till_done() await client.send_json_auto_id(command) result = await client.receive_json() assert result["success"] freezer.move_to(next_time) async_fire_time_changed(hass) await hass.async_block_till_done() assert create_backup.call_count == backup_calls assert get_backups.call_count == get_backups_calls assert delete_backup.call_count == delete_calls assert delete_backup.call_args_list == delete_args_list async_fire_time_changed(hass, fire_all=True) # flush out storage save await hass.async_block_till_done() assert ( hass_storage[DOMAIN]["data"]["config"]["last_attempted_automatic_backup"] == backup_time ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_completed_automatic_backup"] == backup_time ) @pytest.mark.parametrize( ("backup_command", "backup_time"), [ ( {"type": "backup/generate_with_automatic_settings"}, "2024-11-11T12:00:00+01:00", ), ( {"type": "backup/generate", "agent_ids": ["test.test-agent"]}, None, ), ], ) @pytest.mark.parametrize( ( "config_command", "backups", "get_backups_agent_errors", "delete_backup_agent_errors", "backup_calls", "get_backups_calls", "delete_calls", "delete_args_list", ), [ ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": None, "days": None}, "schedule": {"recurrence": "never"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, 1, 1, # we get backups even if backup retention copies is None 0, [], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "never"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, 1, 1, 0, [], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 3, "days": None}, "schedule": {"recurrence": "never"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, 1, 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( { "type": "backup/config/update", "create_backup": {"agent_ids": ["test.test-agent"]}, "retention": {"copies": 2, "days": None}, "schedule": {"recurrence": "never"}, }, { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-5": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-12T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, 1, 1, 2, [ call("backup-1", agent_ids=["test.test-agent"]), call("backup-2", agent_ids=["test.test-agent"]), ], ), ], ) async def test_config_retention_copies_logic_manual_backup( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, hass_storage: dict[str, Any], create_backup: AsyncMock, delete_backup: AsyncMock, get_backups: AsyncMock, config_command: dict[str, Any], backup_command: dict[str, Any], backups: dict[str, Any], get_backups_agent_errors: dict[str, Exception], delete_backup_agent_errors: dict[str, Exception], backup_time: str, backup_calls: int, get_backups_calls: int, delete_calls: int, delete_args_list: Any, ) -> None: """Test config backup retention copies logic for manual backup.""" created_backup: MagicMock = create_backup.return_value[1].result().backup created_backup.protected = True client = await hass_ws_client(hass) storage_data = { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "retention": {"copies": None, "days": None}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": None, "schedule": { "days": [], "recurrence": "daily", "state": "never", "time": None, }, }, } hass_storage[DOMAIN] = { "data": storage_data, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, } get_backups.return_value = (backups, get_backups_agent_errors) delete_backup.return_value = delete_backup_agent_errors await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to("2024-11-11 12:00:00+01:00") await setup_backup_integration(hass, remote_agents=["test-agent"]) await hass.async_block_till_done() await client.send_json_auto_id(config_command) result = await client.receive_json() assert result["success"] # Create a manual backup await client.send_json_auto_id(backup_command) result = await client.receive_json() assert result["success"] # Wait for backup creation to complete await hass.async_block_till_done() assert create_backup.call_count == backup_calls assert get_backups.call_count == get_backups_calls assert delete_backup.call_count == delete_calls assert delete_backup.call_args_list == delete_args_list async_fire_time_changed(hass, fire_all=True) # flush out storage save await hass.async_block_till_done() assert ( hass_storage[DOMAIN]["data"]["config"]["last_attempted_automatic_backup"] == backup_time ) assert ( hass_storage[DOMAIN]["data"]["config"]["last_completed_automatic_backup"] == backup_time ) @pytest.mark.parametrize( ( "stored_retained_days", "commands", "backups", "get_backups_agent_errors", "delete_backup_agent_errors", "last_backup_time", "start_time", "next_time", "get_backups_calls", "delete_calls", "delete_args_list", ), [ # No config update - cleanup backups older than 2 days ( 2, [], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), # No config update - No cleanup ( None, [], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 0, 0, [], ), # Unchanged config ( 2, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 2}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 2}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 3}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 0, [], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 2}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 2, [ call("backup-1", agent_ids=["test.test-agent"]), call("backup-2", agent_ids=["test.test-agent"]), ], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 2}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {"test-agent": BackupAgentError("Boom!")}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 2}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {"test-agent": BackupAgentError("Boom!")}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 1, [call("backup-1", agent_ids=["test.test-agent"])], ), ( None, [ { "type": "backup/config/update", "create_backup": {"agent_ids": ["test-agent"]}, "retention": {"copies": None, "days": 0}, "schedule": {"recurrence": "never"}, } ], { "backup-1": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-09T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-2": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-3": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-11T04:45:00+01:00", with_automatic_settings=True, spec=ManagerBackup, ), "backup-4": MagicMock( agents={"test.test-agent": MagicMock(spec=AgentBackupStatus)}, date="2024-11-10T04:45:00+01:00", with_automatic_settings=False, spec=ManagerBackup, ), }, {}, {}, "2024-11-11T04:45:00+01:00", "2024-11-11T12:00:00+01:00", "2024-11-12T12:00:00+01:00", 1, 2, [ call("backup-1", agent_ids=["test.test-agent"]), call("backup-2", agent_ids=["test.test-agent"]), ], ), ], ) async def test_config_retention_days_logic( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, hass_storage: dict[str, Any], delete_backup: AsyncMock, get_backups: AsyncMock, stored_retained_days: int | None, commands: list[dict[str, Any]], backups: dict[str, Any], get_backups_agent_errors: dict[str, Exception], delete_backup_agent_errors: dict[str, Exception], last_backup_time: str, start_time: str, next_time: str, get_backups_calls: int, delete_calls: int, delete_args_list: list[Any], ) -> None: """Test config backup retention logic.""" client = await hass_ws_client(hass) storage_data = { "backups": [], "config": { "agents": {}, "create_backup": { "agent_ids": ["test-agent"], "include_addons": ["test-addon"], "include_all_addons": False, "include_database": True, "include_folders": ["media"], "name": "test-name", "password": "test-password", }, "retention": {"copies": None, "days": stored_retained_days}, "last_attempted_automatic_backup": None, "last_completed_automatic_backup": last_backup_time, "schedule": { "days": [], "recurrence": "never", "state": "never", "time": None, }, }, } hass_storage[DOMAIN] = { "data": storage_data, "key": DOMAIN, "version": store.STORAGE_VERSION, "minor_version": store.STORAGE_VERSION_MINOR, } get_backups.return_value = (backups, get_backups_agent_errors) delete_backup.return_value = delete_backup_agent_errors await hass.config.async_set_time_zone("Europe/Amsterdam") freezer.move_to(start_time) await setup_backup_integration(hass) await hass.async_block_till_done() for command in commands: await client.send_json_auto_id(command) result = await client.receive_json() assert result["success"] freezer.move_to(next_time) async_fire_time_changed(hass) await hass.async_block_till_done() assert get_backups.call_count == get_backups_calls assert delete_backup.call_count == delete_calls assert delete_backup.call_args_list == delete_args_list async_fire_time_changed(hass, fire_all=True) # flush out storage save await hass.async_block_till_done() async def test_subscribe_event( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, ) -> None: """Test subscribe event.""" await setup_backup_integration(hass, with_hassio=False) manager = hass.data[DATA_MANAGER] client = await hass_ws_client(hass) await client.send_json_auto_id({"type": "backup/subscribe_events"}) assert await client.receive_json() == snapshot assert await client.receive_json() == snapshot manager.async_on_backup_event( CreateBackupEvent(stage=None, state=CreateBackupState.IN_PROGRESS, reason=None) ) assert await client.receive_json() == snapshot @pytest.mark.parametrize( ("agent_id", "backup_id", "password"), [ # Invalid agent or backup ("no_such_agent", "c0cb53bd", "hunter2"), ("backup.local", "no_such_backup", "hunter2"), # Legacy backup, which can't be streamed ("backup.local", "2bcb3113", "hunter2"), # New backup, which can be streamed, try with correct and wrong password ("backup.local", "c0cb53bd", "hunter2"), ("backup.local", "c0cb53bd", "wrong_password"), ], ) @pytest.mark.usefixtures("mock_backups") async def test_can_decrypt_on_download( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, agent_id: str, backup_id: str, password: str, ) -> None: """Test can decrypt on download.""" await setup_backup_integration(hass, with_hassio=False) client = await hass_ws_client(hass) await client.send_json_auto_id( { "type": "backup/can_decrypt_on_download", "backup_id": backup_id, "agent_id": agent_id, "password": password, } ) assert await client.receive_json() == snapshot @pytest.mark.parametrize( "error", [ BackupAgentError, BackupNotFound, ], ) @pytest.mark.usefixtures("mock_backups") async def test_can_decrypt_on_download_with_agent_error( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, error: Exception, ) -> None: """Test can decrypt on download.""" await setup_backup_integration( hass, with_hassio=False, backups={"test.remote": [TEST_BACKUP_ABC123]}, remote_agents=["remote"], ) client = await hass_ws_client(hass) with patch.object(BackupAgentTest, "async_download_backup", side_effect=error): await client.send_json_auto_id( { "type": "backup/can_decrypt_on_download", "backup_id": TEST_BACKUP_ABC123.backup_id, "agent_id": "test.remote", "password": "hunter2", } ) assert await client.receive_json() == snapshot