backups: Add support for creating fully uncompressed backups

supervisor/api/backups.py

@@ -13,6 +13,7 @@
 from ..const import (
     ATTR_ADDONS,
     ATTR_BACKUPS,
+    ATTR_COMPRESSED,
     ATTR_CONTENT,
     ATTR_DATE,
     ATTR_FOLDERS,
@@ -51,6 +52,7 @@
     {
         vol.Optional(ATTR_NAME): str,
         vol.Optional(ATTR_PASSWORD): vol.Maybe(str),
+        vol.Optional(ATTR_COMPRESSED): vol.Maybe(vol.Boolean()),
     }
 )
 
@@ -86,6 +88,7 @@ async def list(self, request):
                     ATTR_TYPE: backup.sys_type,
                     ATTR_SIZE: backup.size,
                     ATTR_PROTECTED: backup.protected,
+                    ATTR_COMPRESSED: backup.compressed,
                     ATTR_CONTENT: {
                         ATTR_HOMEASSISTANT: backup.homeassistant_version is not None,
                         ATTR_ADDONS: backup.addon_list,
@@ -128,6 +131,7 @@ async def info(self, request):
             ATTR_NAME: backup.name,
             ATTR_DATE: backup.date,
             ATTR_SIZE: backup.size,
+            ATTR_COMPRESSED: backup.compressed,
             ATTR_PROTECTED: backup.protected,
             ATTR_HOMEASSISTANT: backup.homeassistant_version,
             ATTR_ADDONS: data_addons,

supervisor/backups/backup.py

@@ -19,6 +19,7 @@
     ATTR_AUDIO_INPUT,
     ATTR_AUDIO_OUTPUT,
     ATTR_BOOT,
+    ATTR_COMPRESSED,
     ATTR_CRYPTO,
     ATTR_DATE,
     ATTR_DOCKER,
@@ -90,6 +91,11 @@ def protected(self):
         """Return backup date."""
         return self._data.get(ATTR_PROTECTED) is not None
 
+    @property
+    def compressed(self):
+        """Return whether backup is compressed."""
+        return self._data.get(ATTR_COMPRESSED)
+
     @property
     def addons(self):
         """Return backup date."""
@@ -152,7 +158,7 @@ def tarfile(self):
         """Return path to backup tarfile."""
         return self._tarfile
 
-    def new(self, slug, name, date, sys_type, password=None):
+    def new(self, slug, name, date, sys_type, password=None, compressed=True):
         """Initialize a new backup."""
         # Init metadata
         self._data[ATTR_SLUG] = slug
@@ -169,6 +175,9 @@ def new(self, slug, name, date, sys_type, password=None):
             self._data[ATTR_PROTECTED] = password_for_validating(password)
             self._data[ATTR_CRYPTO] = CRYPTO_AES128
 
+        if not compressed:
+            self._data[ATTR_COMPRESSED] = False
+
     def set_password(self, password: str) -> bool:
         """Set the password for an existing backup."""
         if not password:
@@ -306,8 +315,9 @@ async def store_addons(self, addon_list: list[str]):
 
         async def _addon_save(addon: Addon):
             """Task to store an add-on into backup."""
+            tar_name = f"{addon.slug}.tar{'.gz' if self.compressed else ''}"
             addon_file = SecureTarFile(
-                Path(self._tmp.name, f"{addon.slug}.tar.gz"), "w", key=self._key
+                Path(self._tmp.name, tar_name), "w", key=self._key, gzip=self.compressed
             )
 
             # Take backup
@@ -340,8 +350,9 @@ async def restore_addons(self, addon_list: list[str]):
 
         async def _addon_restore(addon_slug: str):
             """Task to restore an add-on into backup."""
+            tar_name = f"{addon_slug}.tar{'.gz' if self.compressed else ''}"
             addon_file = SecureTarFile(
-                Path(self._tmp.name, f"{addon_slug}.tar.gz"), "r", key=self._key
+                Path(self._tmp.name, tar_name), "r", key=self._key, gzip=self.compressed
             )
 
             # If exists inside backup
@@ -369,7 +380,9 @@ async def store_folders(self, folder_list: list[str]):
         def _folder_save(name: str):
             """Take backup of a folder."""
             slug_name = name.replace("/", "_")
-            tar_name = Path(self._tmp.name, f"{slug_name}.tar.gz")
+            tar_name = Path(
+                self._tmp.name, f"{slug_name}.tar{'.gz' if self.compressed else ''}"
+            )
             origin_dir = Path(self.sys_config.path_supervisor, name)
 
             # Check if exists
@@ -380,7 +393,9 @@ def _folder_save(name: str):
             # Take backup
             try:
                 _LOGGER.info("Backing up folder %s", name)
-                with SecureTarFile(tar_name, "w", key=self._key) as tar_file:
+                with SecureTarFile(
+                    tar_name, "w", key=self._key, gzip=self.compressed
+                ) as tar_file:
                     atomic_contents_add(
                         tar_file,
                         origin_dir,
@@ -407,7 +422,9 @@ async def restore_folders(self, folder_list: list[str]):
         def _folder_restore(name: str):
             """Intenal function to restore a folder."""
             slug_name = name.replace("/", "_")
-            tar_name = Path(self._tmp.name, f"{slug_name}.tar.gz")
+            tar_name = Path(
+                self._tmp.name, f"{slug_name}.tar{'.gz' if self.compressed else ''}"
+            )
             origin_dir = Path(self.sys_config.path_supervisor, name)
 
             # Check if exists inside backup
@@ -422,7 +439,9 @@ def _folder_restore(name: str):
             # Perform a restore
             try:
                 _LOGGER.info("Restore folder %s", name)
-                with SecureTarFile(tar_name, "r", key=self._key) as tar_file:
+                with SecureTarFile(
+                    tar_name, "r", key=self._key, gzip=self.compressed
+                ) as tar_file:
                     tar_file.extractall(path=origin_dir, members=tar_file)
                 _LOGGER.info("Restore folder %s done", name)
             except (tarfile.TarError, OSError) as err:

supervisor/backups/manager.py

@@ -40,15 +40,17 @@ def get(self, slug):
         """Return backup object."""
         return self._backups.get(slug)
 
-    def _create_backup(self, name, sys_type, password, homeassistant=True):
+    def _create_backup(
+        self, name, sys_type, password, compressed=True, homeassistant=True
+    ):
         """Initialize a new backup object from name."""
         date_str = utcnow().isoformat()
         slug = create_slug(name, date_str)
         tar_file = Path(self.sys_config.path_backup, f"{slug}.tar")
 
         # init object
         backup = Backup(self.coresys, tar_file)
-        backup.new(slug, name, date_str, sys_type, password)
+        backup.new(slug, name, date_str, sys_type, password, compressed)
 
         # set general data
         if homeassistant:
@@ -165,13 +167,13 @@ async def _do_backup(
             self.sys_core.state = CoreState.RUNNING
 
     @Job(conditions=[JobCondition.FREE_SPACE, JobCondition.RUNNING])
-    async def do_backup_full(self, name="", password=None):
+    async def do_backup_full(self, name="", password=None, compressed=True):
         """Create a full backup."""
         if self.lock.locked():
             _LOGGER.error("A backup/restore process is already running")
             return None
 
-        backup = self._create_backup(name, BackupType.FULL, password)
+        backup = self._create_backup(name, BackupType.FULL, password, compressed)
 
         _LOGGER.info("Creating new full backup with slug %s", backup.slug)
         async with self.lock:
@@ -184,7 +186,13 @@ async def do_backup_full(self, name="", password=None):
 
     @Job(conditions=[JobCondition.FREE_SPACE, JobCondition.RUNNING])
     async def do_backup_partial(
-        self, name="", addons=None, folders=None, password=None, homeassistant=True
+        self,
+        name="",
+        addons=None,
+        folders=None,
+        password=None,
+        homeassistant=True,
+        compressed=True,
     ):
         """Create a partial backup."""
         if self.lock.locked():
@@ -197,7 +205,9 @@ async def do_backup_partial(
         if len(addons) == 0 and len(folders) == 0 and not homeassistant:
             _LOGGER.error("Nothing to create backup for")
 
-        backup = self._create_backup(name, BackupType.PARTIAL, password, homeassistant)
+        backup = self._create_backup(
+            name, BackupType.PARTIAL, password, compressed, homeassistant
+        )
 
         _LOGGER.info("Creating new partial backup with slug %s", backup.slug)
         async with self.lock:

supervisor/backups/validate.py

@@ -7,6 +7,7 @@
     ATTR_AUDIO_INPUT,
     ATTR_AUDIO_OUTPUT,
     ATTR_BOOT,
+    ATTR_COMPRESSED,
     ATTR_CRYPTO,
     ATTR_DATE,
     ATTR_DOCKER,
@@ -65,6 +66,7 @@ def unique_addons(addons_list):
         vol.Required(ATTR_TYPE): vol.Coerce(BackupType),
         vol.Required(ATTR_NAME): str,
         vol.Required(ATTR_DATE): str,
+        vol.Optional(ATTR_COMPRESSED, default=True): vol.Boolean(),
         vol.Inclusive(ATTR_PROTECTED, "encrypted"): vol.All(
             str, vol.Length(min=1, max=1)
         ),

supervisor/const.py

@@ -132,6 +132,7 @@
 ATTR_CHASSIS = "chassis"
 ATTR_CHECKS = "checks"
 ATTR_CLI = "cli"
+ATTR_COMPRESSED = "compressed"
 ATTR_CONFIG = "config"
 ATTR_CONFIGURATION = "configuration"
 ATTR_CONNECTED = "connected"

tests/backups/test_manager.py

@@ -23,6 +23,37 @@ async def test_do_backup_full(coresys: CoreSys, backup_mock, install_addon_ssh):
     # Check Backup has been created without password
     assert backup_instance.new.call_args[0][3] == BackupType.FULL
     assert backup_instance.new.call_args[0][4] is None
+    assert backup_instance.new.call_args[0][5] is True
+
+    backup_instance.store_homeassistant.assert_called_once()
+    backup_instance.store_repositories.assert_called_once()
+    backup_instance.store_dockerconfig.assert_called_once()
+
+    backup_instance.store_addons.assert_called_once()
+    assert install_addon_ssh in backup_instance.store_addons.call_args[0][0]
+
+    backup_instance.store_folders.assert_called_once()
+    assert len(backup_instance.store_folders.call_args[0][0]) == 4
+
+    assert coresys.core.state == CoreState.RUNNING
+
+
+async def test_do_backup_full_uncompressed(
+    coresys: CoreSys, backup_mock, install_addon_ssh
+):
+    """Test creating Backup."""
+    coresys.core.state = CoreState.RUNNING
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+
+    manager = BackupManager(coresys)
+
+    # backup_mock fixture causes Backup() to be a MagicMock
+    backup_instance: MagicMock = await manager.do_backup_full(compressed=False)
+
+    # Check Backup has been created without password
+    assert backup_instance.new.call_args[0][3] == BackupType.FULL
+    assert backup_instance.new.call_args[0][4] is None
+    assert backup_instance.new.call_args[0][5] is False
 
     backup_instance.store_homeassistant.assert_called_once()
     backup_instance.store_repositories.assert_called_once()
@@ -53,6 +84,37 @@ async def test_do_backup_partial_minimal(
     # Check Backup has been created without password
     assert backup_instance.new.call_args[0][3] == BackupType.PARTIAL
     assert backup_instance.new.call_args[0][4] is None
+    assert backup_instance.new.call_args[0][5] is True
+
+    backup_instance.store_homeassistant.assert_not_called()
+    backup_instance.store_repositories.assert_called_once()
+    backup_instance.store_dockerconfig.assert_called_once()
+
+    backup_instance.store_addons.assert_not_called()
+
+    backup_instance.store_folders.assert_not_called()
+
+    assert coresys.core.state == CoreState.RUNNING
+
+
+async def test_do_backup_partial_minimal_uncompressed(
+    coresys: CoreSys, backup_mock, install_addon_ssh
+):
+    """Test creating minimal partial Backup."""
+    coresys.core.state = CoreState.RUNNING
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+
+    manager = BackupManager(coresys)
+
+    # backup_mock fixture causes Backup() to be a MagicMock
+    backup_instance: MagicMock = await manager.do_backup_partial(
+        homeassistant=False, compressed=False
+    )
+
+    # Check Backup has been created without password
+    assert backup_instance.new.call_args[0][3] == BackupType.PARTIAL
+    assert backup_instance.new.call_args[0][4] is None
+    assert backup_instance.new.call_args[0][5] is False
 
     backup_instance.store_homeassistant.assert_not_called()
     backup_instance.store_repositories.assert_called_once()
@@ -84,6 +146,7 @@ async def test_do_backup_partial_maximal(
     # Check Backup has been created without password
     assert backup_instance.new.call_args[0][3] == BackupType.PARTIAL
     assert backup_instance.new.call_args[0][4] is None
+    assert backup_instance.new.call_args[0][5] is True
 
     backup_instance.store_homeassistant.assert_called_once()
     backup_instance.store_repositories.assert_called_once()