Handle backups and save data

.vscode/launch.json

@@ -13,6 +13,13 @@
           "remoteRoot": "/usr/src/supervisor"
         }
       ]
+    },
+    {
+      "name": "Debug Tests",
+      "type": "python",
+      "request": "test",
+      "console": "internalConsole",
+      "justMyCode": false
     }
   ]
 }

supervisor/backups/backup.py

@@ -419,7 +419,11 @@ class Backup(CoreSysAttributes):
                 atomic_contents_add(
                     tar_file,
                     origin_dir,
-                    excludes=[],
+                    excludes=[
+                        bound.bind_mount.local_where.as_posix()
+                        for bound in self.sys_mounts.bound_mounts
+                        if bound.bind_mount.local_where
+                    ],
                     arcname=".",
                 )
 

supervisor/mounts/const.py

@@ -1,12 +1,9 @@
 """Constants for mount manager."""
 
 from enum import Enum
-from pathlib import Path
+from pathlib import PurePath
 
-from ..const import SUPERVISOR_DATA
+FILE_CONFIG_MOUNTS = PurePath("mounts.json")
-
-FILE_CONFIG_MOUNTS = Path(SUPERVISOR_DATA, "mounts.json")
-HOST_SUPERVISOR_DATA = Path("/mnt/data/supervisor")
 
 ATTR_MOUNTS = "mounts"
 ATTR_PATH = "path"

supervisor/mounts/manager.py

@@ -33,7 +33,9 @@ class MountManager(FileConfiguration, CoreSysAttributes):
 
     def __init__(self, coresys: CoreSys):
         """Initialize object."""
-        super().__init__(FILE_CONFIG_MOUNTS, SCHEMA_MOUNTS_CONFIG)
+        super().__init__(
+            coresys.config.path_supervisor / FILE_CONFIG_MOUNTS, SCHEMA_MOUNTS_CONFIG
+        )
 
         self.coresys: CoreSys = coresys
         self._mounts: dict[str, Mount] = {
@@ -57,6 +59,11 @@ class MountManager(FileConfiguration, CoreSysAttributes):
         """Return list of media mounts."""
         return [mount for mount in self.mounts if mount.usage == MountUsage.MEDIA]
 
+    @property
+    def bound_mounts(self) -> list[BoundMount]:
+        """Return list of bound mounts and where else they have been bind mounted."""
+        return list(self._bound_mounts.values())
+
     def get(self, name: str) -> Mount:
         """Get mount by name."""
         if name not in self._mounts:
@@ -180,3 +187,10 @@ class MountManager(FileConfiguration, CoreSysAttributes):
             emergency=emergency,
         )
         await bound_mount.bind_mount.load()
+
+    def save_data(self) -> None:
+        """Store data to configuration file."""
+        self._data[ATTR_MOUNTS] = [
+            mount.to_dict(skip_secrets=False) for mount in self.mounts
+        ]
+        super().save_data()

supervisor/mounts/mount.py

@@ -58,7 +58,7 @@ class Mount(CoreSysAttributes, ABC):
 
     def to_dict(self, *, skip_secrets: bool = True) -> MountData:
         """Return dictionary representation."""
-        return MountData(name=self.name, type=self.type, usage=self.usage)
+        return MountData(name=self.name, type=self.type.value, usage=self.usage.value)
 
     @property
     def name(self) -> str:

tests/backups/test_manager.py

@@ -1,15 +1,23 @@
 """Test BackupManager class."""
 
+from shutil import rmtree
 from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch
 
+from dbus_fast import DBusError
+
 from supervisor.addons.addon import Addon
+from supervisor.backups.backup import Backup
 from supervisor.backups.const import BackupType
 from supervisor.backups.manager import BackupManager
 from supervisor.const import FOLDER_HOMEASSISTANT, FOLDER_SHARE, CoreState
 from supervisor.coresys import CoreSys
 from supervisor.exceptions import AddonsError, DockerError
+from supervisor.homeassistant.core import HomeAssistantCore
+from supervisor.mounts.mount import Mount
 
 from tests.const import TEST_ADDON_SLUG
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.systemd import Systemd as SystemdService
 
 
 async def test_do_backup_full(coresys: CoreSys, backup_mock, install_addon_ssh):
@@ -354,3 +362,62 @@ async def test_restore_error(
     await coresys.backups.do_restore_full(backup_instance)
 
     capture_exception.assert_called_once_with(err)
+
+
+async def test_backup_media_with_mounts(
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock],
+    tmp_supervisor_data,
+    path_extern,
+):
+    """Test backing up media folder with mounts."""
+    systemd_service: SystemdService = all_dbus_services["systemd"]
+    systemd_service.response_get_unit = [
+        DBusError("org.freedesktop.systemd1.NoSuchUnit", "error"),
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+        DBusError("org.freedesktop.systemd1.NoSuchUnit", "error"),
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+    ]
+
+    # Make some normal test files
+    (test_file_1 := coresys.config.path_media / "test.txt").touch()
+    (test_dir := coresys.config.path_media / "test").mkdir()
+    (test_file_2 := coresys.config.path_media / "test" / "inner.txt").touch()
+
+    # Add a media mount
+    await coresys.mounts.load()
+    await coresys.mounts.create_mount(
+        Mount.from_dict(
+            coresys,
+            {
+                "name": "media_test",
+                "usage": "media",
+                "type": "cifs",
+                "server": "test.local",
+                "share": "test",
+            },
+        )
+    )
+    assert (mount_dir := coresys.config.path_media / "media_test").is_dir()
+
+    # Make a partial backup
+    coresys.core.state = CoreState.RUNNING
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    backup: Backup = await coresys.backups.do_backup_partial("test", folders=["media"])
+
+    # Remove the mount and wipe the media folder
+    await coresys.mounts.remove_mount("media_test")
+    rmtree(coresys.config.path_media)
+    coresys.config.path_media.mkdir()
+
+    # Restore the backup and check that only the test files we made returned
+    async def mock_async_true(*args, **kwargs):
+        return True
+
+    with patch.object(HomeAssistantCore, "is_running", new=mock_async_true):
+        await coresys.backups.do_restore_partial(backup, folders=["media"])
+
+    assert test_file_1.exists()
+    assert test_dir.is_dir()
+    assert test_file_2.exists()
+    assert not mount_dir.exists()

tests/conftest.py

@@ -370,6 +370,8 @@ async def tmp_supervisor_data(coresys: CoreSys, tmp_path: Path) -> Path:
         coresys.config.path_emergency.mkdir()
         coresys.config.path_media.mkdir()
         coresys.config.path_mounts.mkdir()
+        coresys.config.path_backup.mkdir()
+        coresys.config.path_tmp.mkdir()
         yield tmp_path
 
 

tests/mounts/test_manager.py

@@ -1,6 +1,8 @@
 """Tests for mount manager."""
 
+import json
 import os
+from pathlib import Path
 
 from dbus_fast import DBusError, Variant
 from dbus_fast.aio.message_bus import MessageBus
@@ -9,6 +11,7 @@ import pytest
 from supervisor.coresys import CoreSys
 from supervisor.dbus.const import UnitActiveState
 from supervisor.exceptions import MountNotFound
+from supervisor.mounts.manager import MountManager
 from supervisor.mounts.mount import Mount
 from supervisor.resolution.const import ContextType, IssueType, SuggestionType
 from supervisor.resolution.data import Issue, Suggestion
@@ -360,3 +363,44 @@ async def test_remove_reload_mount_missing(coresys: CoreSys):
 
     with pytest.raises(MountNotFound):
         await coresys.mounts.reload_mount("does_not_exist")
+
+
+async def test_save_data(coresys: CoreSys, tmp_supervisor_data: Path, path_extern):
+    """Test saving mount config data."""
+    # Replace mount manager with one that doesn't have save_data mocked
+    coresys._mounts = MountManager(coresys)  # pylint: disable=protected-access
+
+    path = tmp_supervisor_data / "mounts.json"
+    assert not path.exists()
+
+    await coresys.mounts.load()
+    await coresys.mounts.create_mount(
+        Mount.from_dict(
+            coresys,
+            {
+                "name": "auth_test",
+                "type": "cifs",
+                "usage": "backup",
+                "server": "backup.local",
+                "share": "backups",
+                "username": "admin",
+                "password": "password",
+            },
+        )
+    )
+    coresys.mounts.save_data()
+
+    assert path.exists()
+    with path.open() as f:
+        config = json.load(f)
+        assert config["mounts"] == [
+            {
+                "name": "auth_test",
+                "type": "cifs",
+                "usage": "backup",
+                "server": "backup.local",
+                "share": "backups",
+                "username": "admin",
+                "password": "password",
+            }
+        ]