Use file_filter and add test for addon backup_exclude

supervisor/addons/addon.py

@@ -6,6 +6,7 @@
 from copy import deepcopy
 from datetime import datetime
 import errno
+from functools import partial
 from ipaddress import IPv4Address
 import logging
 from pathlib import Path, PurePath
@@ -1207,6 +1208,25 @@ async def end_backup(self) -> asyncio.Task | None:
             await self._backup_command(self.backup_post)
         return None
 
+    def _is_excluded_by_filter(
+        self, origin_path: Path, arcname: str, item_arcpath: PurePath
+    ) -> bool:
+        """Filter out files from backup based on filters provided by addon developer.
+
+        This tests the dev provided filters against the full path of the file as
+        Supervisor sees them using match. This is done for legacy reasons, testing
+        against the relative path makes more sense and may be changed in the future.
+        """
+        full_path = origin_path / item_arcpath.relative_to(arcname)
+
+        for exclude in self.backup_exclude:
+            if not full_path.match(exclude):
+                continue
+            _LOGGER.debug("Ignoring %s because of %s", full_path, exclude)
+            return True
+
+        return False
+
     @Job(
         name="addon_backup",
         limit=JobExecutionLimit.GROUP_ONCE,
@@ -1266,7 +1286,9 @@ def _write_tarfile():
                     atomic_contents_add(
                         backup,
                         self.path_data,
-                        excludes=self.backup_exclude,
+                        file_filter=partial(
+                            self._is_excluded_by_filter, self.path_data, "data"
+                        ),
                         arcname="data",
                     )
 
@@ -1275,7 +1297,9 @@ def _write_tarfile():
                         atomic_contents_add(
                             backup,
                             self.path_config,
-                            excludes=self.backup_exclude,
+                            file_filter=partial(
+                                self._is_excluded_by_filter, self.path_config, "config"
+                            ),
                             arcname="config",
                         )
 

supervisor/backups/backup.py

@@ -11,7 +11,7 @@
 import io
 import json
 import logging
-from pathlib import Path
+from pathlib import Path, PurePath
 import tarfile
 from tarfile import TarFile
 from tempfile import TemporaryDirectory
@@ -640,6 +640,22 @@ def _save() -> None:
             # Take backup
             _LOGGER.info("Backing up folder %s", name)
 
+            def is_excluded_by_filter(item_arcpath: PurePath) -> bool:
+                """Filter out bind mounts in folders being backed up."""
+                full_path = origin_dir / item_arcpath.relative_to(".")
+
+                for bound in self.sys_mounts.bound_mounts:
+                    if full_path != bound.bind_mount.local_where:
+                        continue
+                    _LOGGER.debug(
+                        "Ignoring %s because of %s",
+                        full_path,
+                        bound.bind_mount.local_where.as_posix(),
+                    )
+                    return True
+
+                return False
+
             with self._outer_secure_tarfile.create_inner_tar(
                 f"./{tar_name}",
                 gzip=self.compressed,
@@ -648,11 +664,7 @@ def _save() -> None:
                 atomic_contents_add(
                     tar_file,
                     origin_dir,
-                    excludes=[
-                        bound.bind_mount.local_where.as_posix()
-                        for bound in self.sys_mounts.bound_mounts
-                        if bound.bind_mount.local_where
-                    ],
+                    file_filter=is_excluded_by_filter,
                     arcname=".",
                 )
 

supervisor/homeassistant/module.py

@@ -416,11 +416,23 @@ def _write_tarfile():
                         if exclude_database:
                             excludes += HOMEASSISTANT_BACKUP_EXCLUDE_DATABASE
 
+                        def is_excluded_by_filter(path: PurePath) -> bool:
+                            """Filter to filter excludes."""
+                            for exclude in excludes:
+                                if not path.match(exclude):
+                                    continue
+                                _LOGGER.debug(
+                                    "Ignoring %s because of %s", path, exclude
+                                )
+                                return True
+
+                            return False
+
                         # Backup data
                         atomic_contents_add(
                             backup,
                             self.sys_config.path_homeassistant,
-                            excludes=excludes,
+                            file_filter=is_excluded_by_filter,
                             arcname="data",
                         )
 

tests/backups/test_manager.py

@@ -2013,3 +2013,29 @@ async def test_backup_remove_one_location_of_multiple(coresys: CoreSys):
     assert not location_2.exists()
     assert coresys.backups.get("7fed74c8")
     assert backup.all_locations == {None: location_1}
+
+
+@pytest.mark.usefixtures("tmp_supervisor_data")
+async def test_addon_backup_excludes(coresys: CoreSys, install_addon_example: Addon):
+    """Test backup excludes option for addons."""
+    coresys.core.state = CoreState.RUNNING
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+
+    install_addon_example.path_data.mkdir(parents=True)
+    (test1 := install_addon_example.path_data / "test1").touch()
+    (test_dir := install_addon_example.path_data / "test_dir").mkdir()
+    (test2 := test_dir / "test2").touch()
+    (test3 := test_dir / "test3").touch()
+
+    install_addon_example.data["backup_exclude"] = ["test1", "*/test2"]
+    backup = await coresys.backups.do_backup_partial(addons=["local_example"])
+    test1.unlink()
+    test2.unlink()
+    test3.unlink()
+    test_dir.rmdir()
+
+    await coresys.backups.do_restore_partial(backup, addons=["local_example"])
+    assert not test1.exists()
+    assert not test2.exists()
+    assert test_dir.is_dir()
+    assert test3.exists()