Extend backup upload API with file name parameter

Add a query parameter which allows to specify the file name on upload.
All locations will store the backup with the same file name.

supervisor/api/backups.py

@@ -26,6 +26,7 @@
     ATTR_DATE,
     ATTR_DAYS_UNTIL_STALE,
     ATTR_EXTRA,
+    ATTR_FILENAME,
     ATTR_FOLDERS,
     ATTR_HOMEASSISTANT,
     ATTR_HOMEASSISTANT_EXCLUDE_DATABASE,
@@ -484,6 +485,10 @@ async def upload(self, request: web.Request):
             if location and location != LOCATION_CLOUD_BACKUP:
                 tmp_path = location.local_where
 
+        filename: str | None = None
+        if ATTR_FILENAME in request.query:
+            filename = request.query.get(ATTR_FILENAME)
+
         with TemporaryDirectory(dir=tmp_path.as_posix()) as temp_dir:
             tar_file = Path(temp_dir, "backup.tar")
             reader = await request.multipart()
@@ -510,7 +515,7 @@ async def upload(self, request: web.Request):
 
             backup = await asyncio.shield(
                 self.sys_backups.import_backup(
-                    tar_file, location=location, additional_locations=locations
+                    tar_file, filename, location=location, additional_locations=locations
                 )
             )
 

supervisor/backups/manager.py

@@ -351,6 +351,7 @@ def copy_to_additional_locations() -> dict[str | None, Path]:
     async def import_backup(
         self,
         tar_file: Path,
+        filename: str | None = None,
         location: LOCATION_TYPE = None,
         additional_locations: list[LOCATION_TYPE] | None = None,
     ) -> Backup | None:
@@ -362,9 +363,13 @@ async def import_backup(
             return None
 
         # Move backup to destination folder
-        tar_origin = Path(self._get_base_path(location), f"{backup.slug}.tar")
+        if filename:
+            tar_file = Path(self._get_base_path(location), Path(filename).name)
+        else:
+            tar_file = Path(self._get_base_path(location), f"{backup.slug}.tar")
+
         try:
-            backup.tarfile.rename(tar_origin)
+            backup.tarfile.rename(tar_file)
 
         except OSError as err:
             if err.errno == errno.EBADMSG and location in {LOCATION_CLOUD_BACKUP, None}:
@@ -373,7 +378,7 @@ async def import_backup(
             return None
 
         # Load new backup
-        backup = Backup(self.coresys, tar_origin, backup.slug, location, backup.data)
+        backup = Backup(self.coresys, tar_file, backup.slug, location, backup.data)
         if not await backup.load():
             # Remove invalid backup from location it was moved to
             backup.tarfile.unlink()

tests/api/test_backups.py

@@ -716,6 +716,29 @@ async def test_upload_duplicate_backup_new_location(
     assert coresys.backups.get("7fed74c8").location is None
 
 
+@pytest.mark.usefixtures("tmp_supervisor_data")
+async def test_upload_with_filename(api_client: TestClient, coresys: CoreSys):
+    """Test uploading a backup to multiple locations."""
+    backup_file = get_fixture_path("backup_example.tar")
+
+    with backup_file.open("rb") as file, MultipartWriter("form-data") as mp:
+        mp.append(file)
+        resp = await api_client.post(
+            "/backups/new/upload?filename=abc.tar", data=mp
+        )
+
+    assert resp.status == 200
+    body = await resp.json()
+    assert body["data"]["slug"] == "7fed74c8"
+
+    orig_backup = coresys.config.path_backup / "abc.tar"
+    assert orig_backup.exists()
+    assert coresys.backups.get("7fed74c8").all_locations == {
+        None: orig_backup,
+    }
+    assert coresys.backups.get("7fed74c8").location is None
+
+
 @pytest.mark.parametrize(
     ("method", "url"),
     [