Add backup config storage (#130871)

* Add base for backup config

* Allow updating backup config

* Test loading backup config

* Add backup config update method

homeassistant/components/backup/config.py

@@ -0,0 +1,77 @@
+"""Provide persistent configuration for the backup integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Self, TypedDict
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.storage import Store
+from homeassistant.helpers.typing import UNDEFINED, UndefinedType
+
+from .const import DOMAIN
+
+STORAGE_KEY = DOMAIN
+STORAGE_VERSION = 1
+
+
+class StoredBackupConfig(TypedDict):
+    """Represent the stored backup config."""
+
+    last_automatic_backup: datetime | None
+    max_copies: int | None
+
+
+@dataclass(kw_only=True)
+class BackupConfigData:
+    """Represent loaded backup config data."""
+
+    last_automatic_backup: datetime | None = None
+    max_copies: int | None = None
+
+    @classmethod
+    def from_dict(cls, data: StoredBackupConfig) -> Self:
+        """Initialize backup config data from a dict."""
+        return cls(
+            last_automatic_backup=data["last_automatic_backup"],
+            max_copies=data["max_copies"],
+        )
+
+    def to_dict(self) -> StoredBackupConfig:
+        """Convert backup config data to a dict."""
+        return StoredBackupConfig(
+            last_automatic_backup=self.last_automatic_backup,
+            max_copies=self.max_copies,
+        )
+
+
+class BackupConfig:
+    """Handle backup config."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize backup config."""
+        self.data = BackupConfigData()
+        self._store: Store[StoredBackupConfig] = Store(
+            hass, STORAGE_VERSION, STORAGE_KEY
+        )
+
+    async def load(self) -> None:
+        """Load config."""
+        stored = await self._store.async_load()
+        if stored:
+            self.data = BackupConfigData.from_dict(stored)
+
+    async def save(self) -> None:
+        """Save config."""
+        await self._store.async_save(self.data.to_dict())
+
+    async def update(
+        self, *, max_copies: int | None | UndefinedType = UNDEFINED
+    ) -> None:
+        """Update config."""
+        for param_name, param_value in {"max_copies": max_copies}.items():
+            if param_value is not UNDEFINED:
+                setattr(self.data, param_name, param_value)
+
+        await self.save()

homeassistant/components/backup/manager.py

@@ -35,6 +35,7 @@ from .agent import (
     BackupAgentPlatformProtocol,
     LocalBackupAgent,
 )
+from .config import BackupConfig
 from .const import (
     BUF_SIZE,
     DOMAIN,
@@ -91,10 +92,12 @@ class BaseBackupManager(abc.ABC, Generic[_BackupT]):
         self.platforms: dict[str, BackupPlatformProtocol] = {}
         self.backup_agents: dict[str, BackupAgent] = {}
         self.local_backup_agents: dict[str, LocalBackupAgent] = {}
+        self.config = BackupConfig(hass)
         self.syncing = False
 
     async def async_setup(self) -> None:
         """Set up the backup manager."""
+        await self.config.load()
         await self.load_platforms()
 
     @callback

homeassistant/components/backup/websocket.py

@@ -29,6 +29,9 @@ def async_register_websocket_handlers(hass: HomeAssistant, with_hassio: bool) ->
     websocket_api.async_register_command(hass, handle_remove)
     websocket_api.async_register_command(hass, handle_restore)
 
+    websocket_api.async_register_command(hass, handle_config_info)
+    websocket_api.async_register_command(hass, handle_config_update)
+
 
 @websocket_api.require_admin
 @websocket_api.websocket_command({vol.Required("type"): "backup/info"})
@@ -269,3 +272,43 @@ async def backup_agents_download(
         return
 
     connection.send_result(msg["id"])
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command({vol.Required("type"): "backup/config/info"})
+@websocket_api.async_response
+async def handle_config_info(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Send the stored backup config."""
+    manager = hass.data[DATA_MANAGER]
+    connection.send_result(
+        msg["id"],
+        {
+            "config": manager.config.data,
+        },
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "backup/config/update",
+        vol.Optional("max_copies"): int,
+    }
+)
+@websocket_api.async_response
+async def handle_config_update(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Update the stored backup config."""
+    manager = hass.data[DATA_MANAGER]
+    changes = dict(msg)
+    changes.pop("id")
+    changes.pop("type")
+    await manager.config.update(**changes)
+    connection.send_result(msg["id"])

tests/components/backup/snapshots/test_websocket.ambr

@@ -257,6 +257,84 @@
     'type': 'result',
   })
 # ---
+# name: test_config_info[storage_data0]
+  dict({
+    'id': 1,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': None,
+        'max_copies': None,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
+# name: test_config_info[storage_data1]
+  dict({
+    'id': 1,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': '2024-10-26T02:00:00+00:00',
+        'max_copies': 3,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
+# name: test_config_info[storage_data2]
+  dict({
+    'id': 1,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': None,
+        'max_copies': 3,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
+# name: test_config_info[storage_data3]
+  dict({
+    'id': 1,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': '2024-10-26T02:00:00+00:00',
+        'max_copies': None,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
+# name: test_config_update
+  dict({
+    'id': 1,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': None,
+        'max_copies': None,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
+# name: test_config_update.1
+  dict({
+    'id': 3,
+    'result': dict({
+      'config': dict({
+        'last_automatic_backup': None,
+        'max_copies': 5,
+      }),
+    }),
+    'success': True,
+    'type': 'result',
+  })
+# ---
 # name: test_details[with_hassio-with_backup_content]
   dict({
     'error': dict({

tests/components/backup/test_websocket.py

@@ -10,7 +10,7 @@ from syrupy import SnapshotAssertion
 
 from homeassistant.components.backup import BaseBackup
 from homeassistant.components.backup.agent import BackupAgentUnreachableError
-from homeassistant.components.backup.const import DATA_MANAGER
+from homeassistant.components.backup.const import DATA_MANAGER, DOMAIN
 from homeassistant.components.backup.manager import NewBackup
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
@@ -539,3 +539,58 @@ async def test_agents_download_unknown_agent(
         }
     )
     assert await client.receive_json() == snapshot
+
+
+@pytest.mark.parametrize(
+    "storage_data",
+    [
+        {},
+        {"last_automatic_backup": "2024-10-26T02:00:00+00:00", "max_copies": 3},
+        {"last_automatic_backup": None, "max_copies": 3},
+        {"last_automatic_backup": "2024-10-26T02:00:00+00:00", "max_copies": None},
+    ],
+)
+async def test_config_info(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    snapshot: SnapshotAssertion,
+    hass_storage: dict[str, Any],
+    storage_data: dict[str, Any],
+) -> None:
+    """Test getting backup config info."""
+    hass_storage[DOMAIN] = {
+        "data": storage_data,
+        "key": DOMAIN,
+        "version": 1,
+    }
+
+    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
+
+
+async def test_config_update(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test 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({"type": "backup/config/update", "max_copies": 5})
+    result = await client.receive_json()
+
+    assert result["success"]
+
+    await client.send_json_auto_id({"type": "backup/config/info"})
+    assert await client.receive_json() == snapshot