Include network storage mount configurations in backups

supervisor/backups/backup.py

@@ -55,6 +55,8 @@
 from ..jobs.const import JOB_GROUP_BACKUP
 from ..jobs.decorator import Job
 from ..jobs.job_group import JobGroup
+from ..mounts.const import ATTR_DEFAULT_BACKUP_MOUNT, ATTR_MOUNTS
+from ..mounts.mount import Mount
 from ..utils import remove_folder
 from ..utils.dt import parse_datetime, utcnow
 from ..utils.json import json_bytes
@@ -208,6 +210,11 @@ def docker(self, value: dict[str, Any]) -> None:
         """Set the Docker config data."""
         self._data[ATTR_DOCKER] = value
 
+    @property
+    def mounts(self) -> dict[str, Any] | None:
+        """Return backup mount configuration data."""
+        return self._data.get(ATTR_MOUNTS)
+
     @property
     def location(self) -> str | None:
         """Return the location of the backup."""
@@ -920,3 +927,69 @@ def restore_repositories(self, replace: bool = False) -> Awaitable[None]:
         return self.sys_store.update_repositories(
             set(self.repositories), issue_on_error=True, replace=replace
         )
+
+    def store_mounts(self) -> None:
+        """Store mount configurations into backup."""
+        mounts_data = {
+            ATTR_DEFAULT_BACKUP_MOUNT: (
+                self.sys_mounts.default_backup_mount.name
+                if self.sys_mounts.default_backup_mount
+                else None
+            ),
+            ATTR_MOUNTS: [
+                mount.to_dict(skip_secrets=False) for mount in self.sys_mounts.mounts
+            ],
+        }
+        self._data[ATTR_MOUNTS] = mounts_data
+
+    async def restore_mounts(self) -> bool:
+        """Restore mount configurations from backup.
+
+        Returns True if all mounts were restored successfully.
+        """
+        if not self.mounts:
+            return True
+
+        success = True
+        restored_mounts: list[str] = []
+
+        # Restore each mount configuration
+        for mount_data in self.mounts.get(ATTR_MOUNTS, []):
+            mount_name = mount_data.get("name", "unknown")
+            try:
+                # Skip if mount already exists
+                if mount_name in self.sys_mounts:
+                    _LOGGER.info(
+                        "Mount %s already exists, skipping restore", mount_name
+                    )
+                    continue
+
+                # Create mount from backup data
+                mount = Mount.from_dict(self.coresys, mount_data)
+                await self.sys_mounts.create_mount(mount)
+                restored_mounts.append(mount_name)
+                _LOGGER.info("Restored mount configuration: %s", mount_name)
+            except Exception as err:  # pylint: disable=broad-except
+                _LOGGER.warning("Failed to restore mount %s: %s", mount_name, err)
+                success = False
+
+        # Restore default backup mount if specified and mount exists
+        default_mount_name = self.mounts.get(ATTR_DEFAULT_BACKUP_MOUNT)
+        if default_mount_name and default_mount_name in self.sys_mounts:
+            try:
+                self.sys_mounts.default_backup_mount = self.sys_mounts.get(
+                    default_mount_name
+                )
+                _LOGGER.info("Restored default backup mount: %s", default_mount_name)
+            except Exception as err:  # pylint: disable=broad-except
+                _LOGGER.warning(
+                    "Failed to restore default backup mount %s: %s",
+                    default_mount_name,
+                    err,
+                )
+
+        # Save mount configuration
+        if restored_mounts:
+            await self.sys_mounts.save_data()
+
+        return success

supervisor/backups/const.py

@@ -27,6 +27,7 @@ class BackupJobStage(StrEnum):
     FINISHING_FILE = "finishing_file"
     FOLDERS = "folders"
     HOME_ASSISTANT = "home_assistant"
+    MOUNTS = "mounts"
     COPY_ADDITONAL_LOCATIONS = "copy_additional_locations"
     AWAIT_ADDON_RESTARTS = "await_addon_restarts"
 
@@ -40,4 +41,5 @@ class RestoreJobStage(StrEnum):
     AWAIT_HOME_ASSISTANT_RESTART = "await_home_assistant_restart"
     FOLDERS = "folders"
     HOME_ASSISTANT = "home_assistant"
+    MOUNTS = "mounts"
     REMOVE_DELTA_ADDONS = "remove_delta_addons"

supervisor/backups/manager.py

@@ -549,6 +549,10 @@ async def _do_backup(
                     self._change_stage(BackupJobStage.FOLDERS, backup)
                     await backup.store_folders(folder_list)
 
+                # Backup mount configurations
+                self._change_stage(BackupJobStage.MOUNTS, backup)
+                backup.store_mounts()
+
                 self._change_stage(BackupJobStage.FINISHING_FILE, backup)
 
         except BackupError as err:
@@ -750,6 +754,12 @@ async def _do_restore(
                     )
                     success = success and restore_success
 
+                # Restore mount configurations
+                if backup.mounts:
+                    self._change_stage(RestoreJobStage.MOUNTS, backup)
+                    mount_success = await backup.restore_mounts()
+                    success = success and mount_success
+
                 # Wait for Home Assistant Core update/downgrade
                 if task_hass:
                     await task_hass

supervisor/backups/validate.py

@@ -20,12 +20,15 @@
     ATTR_FOLDERS,
     ATTR_HOMEASSISTANT,
     ATTR_NAME,
+    ATTR_PASSWORD,
+    ATTR_PORT,
     ATTR_PROTECTED,
     ATTR_REPOSITORIES,
     ATTR_SIZE,
     ATTR_SLUG,
     ATTR_SUPERVISOR_VERSION,
     ATTR_TYPE,
+    ATTR_USERNAME,
     ATTR_VERSION,
     CRYPTO_AES128,
     FOLDER_ADDONS,
@@ -34,6 +37,18 @@
     FOLDER_SHARE,
     FOLDER_SSL,
 )
+from ..mounts.const import (
+    ATTR_DEFAULT_BACKUP_MOUNT,
+    ATTR_MOUNTS,
+    ATTR_PATH,
+    ATTR_READ_ONLY,
+    ATTR_SERVER,
+    ATTR_SHARE,
+    ATTR_USAGE,
+    MountCifsVersion,
+    MountType,
+    MountUsage,
+)
 from ..store.validate import repositories
 from ..validate import SCHEMA_DOCKER_CONFIG, version_tag
 
@@ -134,6 +149,43 @@ def v1_protected(protected: bool | str) -> bool:
         ),
         vol.Optional(ATTR_REPOSITORIES, default=list): repositories,
         vol.Optional(ATTR_EXTRA, default=dict): dict,
+        vol.Optional(ATTR_MOUNTS, default=None): vol.Maybe(
+            vol.Schema(
+                {
+                    vol.Optional(ATTR_DEFAULT_BACKUP_MOUNT, default=None): vol.Maybe(
+                        str
+                    ),
+                    vol.Required(ATTR_MOUNTS, default=list): [
+                        vol.Schema(
+                            {
+                                vol.Required(ATTR_NAME): str,
+                                vol.Required(ATTR_TYPE): vol.Coerce(MountType),
+                                vol.Required(ATTR_USAGE): vol.Maybe(
+                                    vol.Coerce(MountUsage)
+                                ),
+                                vol.Optional(ATTR_READ_ONLY, default=False): (
+                                    vol.Boolean()
+                                ),
+                                # Network mount fields
+                                vol.Optional(ATTR_SERVER): str,
+                                vol.Optional(ATTR_PORT): int,
+                                # CIFS fields
+                                vol.Optional(ATTR_SHARE): str,
+                                vol.Optional(ATTR_USERNAME): str,
+                                vol.Optional(ATTR_PASSWORD): str,
+                                vol.Optional(ATTR_VERSION): vol.Maybe(
+                                    vol.Coerce(MountCifsVersion)
+                                ),
+                                # NFS/Bind fields
+                                vol.Optional(ATTR_PATH): str,
+                            },
+                            extra=vol.REMOVE_EXTRA,
+                        )
+                    ],
+                },
+                extra=vol.REMOVE_EXTRA,
+            )
+        ),
     },
     extra=vol.ALLOW_EXTRA,
 )

tests/backups/conftest.py

@@ -25,11 +25,14 @@ def fixture_backup_mock():
         backup_instance.store_folders = AsyncMock(return_value=None)
         backup_instance.store_homeassistant = AsyncMock(return_value=None)
         backup_instance.store_addons = AsyncMock(return_value=None)
+        backup_instance.store_mounts = MagicMock(return_value=None)
         backup_instance.restore_folders = AsyncMock(return_value=True)
         backup_instance.restore_homeassistant = AsyncMock(return_value=None)
         backup_instance.restore_addons = AsyncMock(return_value=(True, []))
         backup_instance.restore_repositories = AsyncMock(return_value=None)
+        backup_instance.restore_mounts = AsyncMock(return_value=True)
         backup_instance.remove_delta_addons = AsyncMock(return_value=True)
+        backup_instance.mounts = None
 
         yield backup_mock
 

tests/backups/test_backup.py

@@ -22,6 +22,7 @@
     BackupPermissionError,
 )
 from supervisor.jobs import JobSchedulerOptions
+from supervisor.mounts.mount import Mount
 
 from tests.common import get_fixture_path
 
@@ -273,3 +274,71 @@ async def test_validate_backup(
         expected_exception,
     ):
         await enc_backup.validate_backup(None)
+
+
+async def test_store_mounts(coresys: CoreSys, tmp_path: Path, mount_propagation):
+    """Test storing mount configurations in backup."""
+    backup = Backup(coresys, tmp_path / "my_backup.tar", "test", None)
+    backup.new("test", "2023-07-21T21:05:00.000000+00:00", BackupType.FULL)
+
+    # Initially no mounts
+    assert backup.mounts is None
+
+    # Store mounts (empty list when no mounts configured)
+    backup.store_mounts()
+
+    # Verify mounts data is stored
+    assert backup.mounts is not None
+    assert "mounts" in backup.mounts
+    assert "default_backup_mount" in backup.mounts
+    assert backup.mounts["default_backup_mount"] is None
+    assert backup.mounts["mounts"] == []
+
+
+async def test_store_mounts_with_configured_mounts(
+    coresys: CoreSys, tmp_path: Path, mount_propagation, mock_is_mount
+):
+    """Test storing mount configurations when mounts are configured."""
+    # Load mounts system
+    await coresys.mounts.load()
+
+    # Create a test mount
+    mount = Mount.from_dict(
+        coresys,
+        {
+            "name": "test_backup_share",
+            "usage": "backup",
+            "type": "cifs",
+            "server": "192.168.1.100",
+            "share": "backup_share",
+        },
+    )
+    await coresys.mounts.create_mount(mount)
+
+    backup = Backup(coresys, tmp_path / "my_backup.tar", "test", None)
+    backup.new("test", "2023-07-21T21:05:00.000000+00:00", BackupType.FULL)
+
+    # Store mounts
+    backup.store_mounts()
+
+    # Verify mount data is stored
+    assert backup.mounts is not None
+    assert len(backup.mounts["mounts"]) == 1
+    stored_mount = backup.mounts["mounts"][0]
+    assert stored_mount["name"] == "test_backup_share"
+    assert stored_mount["type"] == "cifs"
+    assert stored_mount["server"] == "192.168.1.100"
+    assert stored_mount["share"] == "backup_share"
+
+
+async def test_restore_mounts_empty(coresys: CoreSys, tmp_path: Path):
+    """Test restoring mounts when backup has no mounts."""
+    backup = Backup(coresys, tmp_path / "my_backup.tar", "test", None)
+    backup.new("test", "2023-07-21T21:05:00.000000+00:00", BackupType.FULL)
+
+    # No mounts in backup
+    assert backup.mounts is None
+
+    # Restore should succeed with nothing to do
+    success = await backup.restore_mounts()
+    assert success is True