Support share mounts (#4318)

supervisor/mounts/const.py

@@ -26,3 +26,4 @@ class MountUsage(str, Enum):
 
     BACKUP = "backup"
     MEDIA = "media"
+    SHARE = "share"

supervisor/mounts/manager.py

@@ -68,6 +68,11 @@ class MountManager(FileConfiguration, CoreSysAttributes):
         """Return list of media mounts."""
         return [mount for mount in self.mounts if mount.usage == MountUsage.MEDIA]
 
+    @property
+    def share_mounts(self) -> list[Mount]:
+        """Return list of share mounts."""
+        return [mount for mount in self.mounts if mount.usage == MountUsage.SHARE]
+
     @property
     def bound_mounts(self) -> list[BoundMount]:
         """Return list of bound mounts and where else they have been bind mounted."""
@@ -125,6 +130,15 @@ class MountManager(FileConfiguration, CoreSysAttributes):
                 ]
             )
 
+        # Bind all share mounts to directories in share
+        if self.share_mounts:
+            await asyncio.wait(
+                [
+                    self.sys_create_task(self._bind_share(mount))
+                    for mount in self.share_mounts
+                ]
+            )
+
     @Job(conditions=[JobCondition.MOUNT_AVAILABLE])
     async def reload(self) -> None:
         """Update mounts info via dbus and reload failed mounts."""
@@ -180,6 +194,8 @@ class MountManager(FileConfiguration, CoreSysAttributes):
         self._mounts[mount.name] = mount
         if mount.usage == MountUsage.MEDIA:
             await self._bind_media(mount)
+        elif mount.usage == MountUsage.SHARE:
+            await self._bind_share(mount)
 
     @Job(conditions=[JobCondition.MOUNT_AVAILABLE], on_condition=MountJobError)
     async def remove_mount(self, name: str, *, retain_entry: bool = False) -> None:
@@ -222,6 +238,10 @@ class MountManager(FileConfiguration, CoreSysAttributes):
         """Bind a media mount to media directory."""
         await self._bind_mount(mount, self.sys_config.path_extern_media / mount.name)
 
+    async def _bind_share(self, mount: Mount) -> None:
+        """Bind a share mount to share directory."""
+        await self._bind_mount(mount, self.sys_config.path_extern_share / mount.name)
+
     async def _bind_mount(self, mount: Mount, where: PurePath) -> None:
         """Bind mount to path, falling back on emergency if necessary.
 

tests/backups/test_manager.py

@@ -427,6 +427,67 @@ async def test_backup_media_with_mounts(
     assert not mount_dir.exists()
 
 
+async def test_backup_share_with_mounts(
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock],
+    tmp_supervisor_data,
+    path_extern,
+):
+    """Test backing up share 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",
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+    ]
+
+    # Make some normal test files
+    (test_file_1 := coresys.config.path_share / "test.txt").touch()
+    (test_dir := coresys.config.path_share / "test").mkdir()
+    (test_file_2 := coresys.config.path_share / "test" / "inner.txt").touch()
+
+    # Add a media mount
+    await coresys.mounts.load()
+    await coresys.mounts.create_mount(
+        Mount.from_dict(
+            coresys,
+            {
+                "name": "share_test",
+                "usage": "share",
+                "type": "cifs",
+                "server": "test.local",
+                "share": "test",
+            },
+        )
+    )
+    assert (mount_dir := coresys.config.path_share / "share_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=["share"])
+
+    # Remove the mount and wipe the media folder
+    await coresys.mounts.remove_mount("share_test")
+    rmtree(coresys.config.path_share)
+    coresys.config.path_share.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=["share"])
+
+    assert test_file_1.exists()
+    assert test_dir.is_dir()
+    assert test_file_2.exists()
+    assert not mount_dir.exists()
+
+
 async def test_full_backup_to_mount(coresys: CoreSys, tmp_supervisor_data, path_extern):
     """Test full backup to and restoring from a mount."""
     (marker := coresys.config.path_homeassistant / "test.txt").touch()

tests/conftest.py

@@ -376,6 +376,7 @@ async def tmp_supervisor_data(coresys: CoreSys, tmp_path: Path) -> Path:
         coresys.config.path_homeassistant.mkdir()
         coresys.config.path_audio.mkdir()
         coresys.config.path_dns.mkdir()
+        coresys.config.path_share.mkdir()
         yield tmp_path
 
 

tests/mounts/test_manager.py

@@ -41,6 +41,13 @@ MEDIA_TEST_DATA = {
     "server": "media.local",
     "path": "/media",
 }
+SHARE_TEST_DATA = {
+    "name": "share_test",
+    "type": "nfs",
+    "usage": "share",
+    "server": "share.local",
+    "path": "/share",
+}
 
 
 @pytest.fixture(name="mount")
@@ -135,6 +142,68 @@ async def test_load(
     ]
 
 
+async def test_load_share_mount(
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock],
+    tmp_supervisor_data,
+    path_extern,
+):
+    """Test mount manager loading with share mount."""
+    systemd_service: SystemdService = all_dbus_services["systemd"]
+    systemd_service.StartTransientUnit.calls.clear()
+
+    share_test = Mount.from_dict(coresys, SHARE_TEST_DATA)
+    # pylint: disable=protected-access
+    coresys.mounts._mounts = {
+        "share_test": share_test,
+    }
+    # pylint: enable=protected-access
+    assert coresys.mounts.share_mounts == [share_test]
+
+    assert share_test.state is None
+    assert not share_test.local_where.exists()
+    assert not any(coresys.config.path_share.iterdir())
+
+    systemd_service.response_get_unit = {
+        "mnt-data-supervisor-mounts-share_test.mount": [
+            ERROR_NO_UNIT,
+            "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+        ],
+        "mnt-data-supervisor-share-share_test.mount": [
+            ERROR_NO_UNIT,
+            "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+        ],
+    }
+    await coresys.mounts.load()
+
+    assert share_test.state == UnitActiveState.ACTIVE
+    assert share_test.local_where.is_dir()
+    assert (coresys.config.path_share / "share_test").is_dir()
+
+    assert systemd_service.StartTransientUnit.calls == [
+        (
+            "mnt-data-supervisor-mounts-share_test.mount",
+            "fail",
+            [
+                ["Type", Variant("s", "nfs")],
+                ["Description", Variant("s", "Supervisor nfs mount: share_test")],
+                ["What", Variant("s", "share.local:/share")],
+            ],
+            [],
+        ),
+        (
+            "mnt-data-supervisor-share-share_test.mount",
+            "fail",
+            [
+                ["Options", Variant("s", "bind")],
+                ["Description", Variant("s", "Supervisor bind mount: bind_share_test")],
+                ["What", Variant("s", "/mnt/data/supervisor/mounts/share_test")],
+            ],
+            [],
+        ),
+    ]
+
+
 async def test_mount_failed_during_load(
     coresys: CoreSys,
     all_dbus_services: dict[str, DBusServiceMock],
@@ -541,3 +610,44 @@ async def test_mounting_not_supported(
 
     with pytest.raises(MountJobError):
         await coresys.mounts.remove_mount("media_test")
+
+
+async def test_create_share_mount(
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock],
+    tmp_supervisor_data,
+    path_extern,
+):
+    """Test creating a share mount."""
+    systemd_service: SystemdService = all_dbus_services["systemd"]
+    systemd_service.StartTransientUnit.calls.clear()
+
+    await coresys.mounts.load()
+
+    mount = Mount.from_dict(coresys, SHARE_TEST_DATA)
+
+    assert mount.state is None
+    assert mount not in coresys.mounts
+    assert "share_test" not in coresys.mounts
+    assert not mount.local_where.exists()
+    assert not any(coresys.config.path_share.iterdir())
+
+    # Create the mount
+    systemd_service.response_get_unit = [
+        ERROR_NO_UNIT,
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+        ERROR_NO_UNIT,
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
+    ]
+    await coresys.mounts.create_mount(mount)
+
+    assert mount.state == UnitActiveState.ACTIVE
+    assert mount in coresys.mounts
+    assert "share_test" in coresys.mounts
+    assert mount.local_where.exists()
+    assert (coresys.config.path_share / "share_test").exists()
+
+    assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
+        "mnt-data-supervisor-mounts-share_test.mount",
+        "mnt-data-supervisor-share-share_test.mount",
+    ]