Fix backup consolidate and upload duplicate (#5472)

Author: mdegat01

supervisor/backups/manager.py

@@ -359,7 +359,7 @@ async def import_backup(
             return None
 
         # Load new backup
-        backup = Backup(self.coresys, tar_origin, backup.slug, None, backup.data)
+        backup = Backup(self.coresys, tar_origin, backup.slug, location, backup.data)
         if not await backup.load():
             # Remove invalid backup from location it was moved to
             backup.tarfile.unlink()
@@ -369,7 +369,7 @@ async def import_backup(
         # Already exists?
         if (
             backup.slug in self._backups
-            and backup.all_locations != self._backups[backup].all_locations
+            and backup.all_locations != self._backups[backup.slug].all_locations
         ):
             _LOGGER.warning("Backup %s already exists! consolidating", backup.slug)
             try:

tests/api/test_backups.py

@@ -698,3 +698,34 @@ async def test_upload_to_multiple_locations(
         ".cloud_backup": copy_backup,
     }
     assert coresys.backups.get("7fed74c8").location is None
+
+
+async def test_upload_duplicate_backup_new_location(
+    api_client: TestClient,
+    coresys: CoreSys,
+    tmp_supervisor_data: Path,
+):
+    """Test uploading a backup that already exists to a new location."""
+    backup_file = get_fixture_path("backup_example.tar")
+    orig_backup = Path(copy(backup_file, coresys.config.path_backup))
+    await coresys.backups.reload(None, "backup_example.tar")
+    assert coresys.backups.get("7fed74c8").all_locations == {None: orig_backup}
+
+    with backup_file.open("rb") as file, MultipartWriter("form-data") as mp:
+        mp.append(file)
+        resp = await api_client.post(
+            "/backups/new/upload?location=.cloud_backup", data=mp
+        )
+
+    assert resp.status == 200
+    body = await resp.json()
+    assert body["data"]["slug"] == "7fed74c8"
+
+    copy_backup = coresys.config.path_core_backup / "7fed74c8.tar"
+    assert orig_backup.exists()
+    assert copy_backup.exists()
+    assert coresys.backups.get("7fed74c8").all_locations == {
+        None: orig_backup,
+        ".cloud_backup": copy_backup,
+    }
+    assert coresys.backups.get("7fed74c8").location is None