Test cases and small fixes identified

supervisor/api/backups.py

@@ -18,7 +18,6 @@
 from ..backups.validate import ALL_FOLDERS, FOLDER_HOMEASSISTANT, days_until_stale
 from ..const import (
     ATTR_ADDONS,
-    ATTR_BACKUP,
     ATTR_BACKUPS,
     ATTR_COMPRESSED,
     ATTR_CONTENT,
@@ -55,7 +54,7 @@
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
 RE_SLUGIFY_NAME = re.compile(r"[^A-Za-z0-9]+")
-RE_FILENAME = re.compile(r"^[^\\\/]+")
+RE_BACKUP_FILENAME = re.compile(r"^[^\\\/]+\.tar$")
 
 # Backwards compatible
 # Remove: 2022.08
@@ -109,12 +108,8 @@
 )
 SCHEMA_RELOAD = vol.Schema(
     {
-        vol.Optional(ATTR_BACKUP, default={}): vol.Schema(
-            {
-                vol.Required(ATTR_LOCATION): vol.Maybe(str),
-                vol.Required(ATTR_FILENAME): vol.Match(RE_FILENAME),
-            }
-        )
+        vol.Inclusive(ATTR_LOCATION, "file"): vol.Maybe(str),
+        vol.Inclusive(ATTR_FILENAME, "file"): vol.Match(RE_BACKUP_FILENAME),
     }
 )
 
@@ -185,7 +180,8 @@ async def options(self, request):
     async def reload(self, request: web.Request):
         """Reload backup list."""
         body = await api_validate(SCHEMA_RELOAD, request)
-        backup = self._location_to_mount(body[ATTR_BACKUP])
+        self._validate_cloud_backup_location(request, body.get(ATTR_LOCATION))
+        backup = self._location_to_mount(body)
 
         return await asyncio.shield(self.sys_backups.reload(**backup))
 

supervisor/api/const.py

@@ -68,6 +68,7 @@
 ATTR_USAGE = "usage"
 ATTR_USE_NTP = "use_ntp"
 ATTR_USERS = "users"
+ATTR_USER_PATH = "user_path"
 ATTR_VENDOR = "vendor"
 ATTR_VIRTUALIZATION = "virtualization"
 

supervisor/api/mounts.py

@@ -11,7 +11,7 @@
 from ..mounts.const import ATTR_DEFAULT_BACKUP_MOUNT, MountUsage
 from ..mounts.mount import Mount
 from ..mounts.validate import SCHEMA_MOUNT_CONFIG
-from .const import ATTR_MOUNTS
+from .const import ATTR_MOUNTS, ATTR_USER_PATH
 from .utils import api_process, api_validate
 
 SCHEMA_OPTIONS = vol.Schema(
@@ -32,7 +32,11 @@ async def info(self, request: web.Request) -> dict[str, Any]:
             if self.sys_mounts.default_backup_mount
             else None,
             ATTR_MOUNTS: [
-                mount.to_dict() | {ATTR_STATE: mount.state}
+                mount.to_dict()
+                | {
+                    ATTR_STATE: mount.state,
+                    ATTR_USER_PATH: mount.container_where.as_posix(),
+                }
                 for mount in self.sys_mounts.mounts
             ],
         }

supervisor/backups/backup.py

@@ -212,6 +212,9 @@ def container_path(self) -> PurePath | None:
         path_map: dict[Path, PurePath] = {
             self.sys_config.path_backup: PATH_BACKUP,
             self.sys_config.path_core_backup: PATH_CLOUD_BACKUP,
+        } | {
+            mount.local_where: mount.container_where
+            for mount in self.sys_mounts.backup_mounts
         }
         for source, target in path_map.items():
             if self.tarfile.is_relative_to(source):

supervisor/backups/manager.py

@@ -14,7 +14,7 @@
     ATTR_DATA,
     ATTR_DAYS_UNTIL_STALE,
     ATTR_JOB_ID,
-    ATTR_LOCATION,
+    ATTR_PATH,
     ATTR_SLUG,
     ATTR_TYPE,
     FILE_HASSIO_BACKUPS,
@@ -84,11 +84,10 @@ def backup_locations(self) -> dict[str | None, Path]:
         return {
             None: self.sys_config.path_backup,
             LOCATION_CLOUD_BACKUP: self.sys_config.path_core_backup,
-            **{
-                mount.name: mount.local_where
-                for mount in self.sys_mounts.backup_mounts
-                if mount.state == UnitActiveState.ACTIVE
-            },
+        } | {
+            mount.name: mount.local_where
+            for mount in self.sys_mounts.backup_mounts
+            if mount.state == UnitActiveState.ACTIVE
         }
 
     def get(self, slug: str) -> Backup:
@@ -241,7 +240,8 @@ async def _load_backup(location: str | None, tar_file: Path) -> bool:
 
         if location != DEFAULT and filename:
             return await _load_backup(
-                location, self._get_base_path(location) / filename
+                self._get_location_name(location),
+                self._get_base_path(location) / filename,
             )
 
         self._backups = {}
@@ -358,13 +358,13 @@ async def _do_backup(
             return None
         else:
             self._backups[backup.slug] = backup
-            self.sys_homeassistant.websocket.async_send_message(
+            await self.sys_homeassistant.websocket.async_send_message(
                 {
                     ATTR_TYPE: WSType.BACKUP_COMPLETE,
                     ATTR_DATA: {
                         ATTR_JOB_ID: self.sys_jobs.current.uuid,
                         ATTR_SLUG: backup.slug,
-                        ATTR_LOCATION: backup.location,
+                        ATTR_PATH: backup.container_path.as_posix(),
                     },
                 }
             )

supervisor/const.py

@@ -260,6 +260,7 @@
 ATTR_PANELS = "panels"
 ATTR_PARENT = "parent"
 ATTR_PASSWORD = "password"
+ATTR_PATH = "path"
 ATTR_PLUGINS = "plugins"
 ATTR_PORT = "port"
 ATTR_PORTS = "ports"

supervisor/docker/homeassistant.py

@@ -135,7 +135,7 @@ def mounts(self) -> list[Mount]:
                     Mount(
                         type=MountType.BIND,
                         source=self.sys_config.path_extern_backup.as_posix(),
-                        target=PATH_BACKUP,
+                        target=PATH_BACKUP.as_posix(),
                         read_only=False,
                         propagation=PropagationMode.RSLAVE.value,
                     ),

supervisor/mounts/mount.py

@@ -164,24 +164,19 @@ def local_where(self) -> Path | None:
             else None
         )
 
-    @cached_property
-    def container_where(self) -> Path | None:
+    @property
+    def container_where(self) -> PurePath | None:
         """Return where this is made available in managed containers (core, addons, etc.).
 
-        This returns none if 'local_where' is none or not a place mapped into other containers.
+        This returns none if it is not made available in managed containers.
         """
-        if not (local_where := self.local_where):
-            return None
-
-        path_map: dict[Path, PurePath] = {
-            self.sys_config.path_backup: PATH_BACKUP,
-            self.sys_config.path_media: PATH_MEDIA,
-            self.sys_config.path_share: PATH_SHARE,
-        }
-        for source, target in path_map.items():
-            if local_where.is_relative_to(source):
-                return target / local_where.relative_to(source)
-
+        match self.usage:
+            case MountUsage.BACKUP:
+                return PurePath(PATH_BACKUP, self.name)
+            case MountUsage.MEDIA:
+                return PurePath(PATH_MEDIA, self.name)
+            case MountUsage.SHARE:
+                return PurePath(PATH_SHARE, self.name)
         return None
 
     @property

tests/api/test_backups.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from pathlib import Path, PurePath
+from shutil import copy
 from typing import Any
 from unittest.mock import ANY, AsyncMock, PropertyMock, patch
 
@@ -19,6 +20,9 @@
 from supervisor.mounts.mount import Mount
 from supervisor.supervisor import Supervisor
 
+from tests.common import get_fixture_path
+from tests.const import TEST_ADDON_SLUG
+
 
 async def test_info(api_client, coresys: CoreSys, mock_full_backup: Backup):
     """Test info endpoint."""
@@ -467,3 +471,120 @@ async def test_restore_immediate_errors(
         )
     assert resp.status == 400
     assert "No Home Assistant" in (await resp.json())["message"]
+
+
+@pytest.mark.parametrize(
+    ("folder", "location"), [("backup", None), ("core/backup", ".cloud_backup")]
+)
+async def test_reload(
+    request: pytest.FixtureRequest,
+    api_client: TestClient,
+    coresys: CoreSys,
+    tmp_supervisor_data: Path,
+    folder: str,
+    location: str | None,
+):
+    """Test backups reload."""
+    assert not coresys.backups.list_backups
+
+    backup_file = get_fixture_path("backup_example.tar")
+    copy(backup_file, tmp_supervisor_data / folder)
+
+    resp = await api_client.post("/backups/reload")
+    assert resp.status == 200
+
+    assert len(coresys.backups.list_backups) == 1
+    assert (backup := coresys.backups.get("7fed74c8"))
+    assert backup.location == location
+    assert backup.locations == [location]
+
+
+@pytest.mark.parametrize(
+    ("folder", "location"), [("backup", None), ("core/backup", ".cloud_backup")]
+)
+async def test_partial_reload(
+    request: pytest.FixtureRequest,
+    api_client: TestClient,
+    coresys: CoreSys,
+    tmp_supervisor_data: Path,
+    folder: str,
+    location: str | None,
+):
+    """Test partial backups reload."""
+    assert not coresys.backups.list_backups
+
+    backup_file = get_fixture_path("backup_example.tar")
+    copy(backup_file, tmp_supervisor_data / folder)
+
+    resp = await api_client.post(
+        "/backups/reload", json={"location": location, "filename": "backup_example.tar"}
+    )
+    assert resp.status == 200
+
+    assert len(coresys.backups.list_backups) == 1
+    assert (backup := coresys.backups.get("7fed74c8"))
+    assert backup.location == location
+    assert backup.locations == [location]
+
+
+async def test_invalid_reload(api_client: TestClient):
+    """Test invalid reload."""
+    resp = await api_client.post("/backups/reload", json={"location": "no_filename"})
+    assert resp.status == 400
+
+    resp = await api_client.post(
+        "/backups/reload", json={"filename": "no_location.tar"}
+    )
+    assert resp.status == 400
+
+    resp = await api_client.post(
+        "/backups/reload", json={"location": None, "filename": "no/sub/paths.tar"}
+    )
+    assert resp.status == 400
+
+    resp = await api_client.post(
+        "/backups/reload", json={"location": None, "filename": "not_tar.tar.gz"}
+    )
+    assert resp.status == 400
+
+
+@pytest.mark.usefixtures("install_addon_ssh")
+@pytest.mark.parametrize("api_client", TEST_ADDON_SLUG, indirect=True)
+async def test_cloud_backup_core_only(api_client: TestClient, mock_full_backup: Backup):
+    """Test only core can access cloud backup location."""
+    resp = await api_client.post(
+        "/backups/reload",
+        json={"location": ".cloud_backup", "filename": "caller_not_core.tar"},
+    )
+    assert resp.status == 403
+
+    resp = await api_client.post(
+        "/backups/new/full",
+        json={
+            "name": "Mount test",
+            "location": ".cloud_backup",
+        },
+    )
+    assert resp.status == 403
+
+    resp = await api_client.post(
+        "/backups/new/partial",
+        json={"name": "Test", "homeassistant": True, "location": ".cloud_backup"},
+    )
+    assert resp.status == 403
+
+    # pylint: disable-next=protected-access
+    mock_full_backup._locations = {".cloud_backup": None}
+    assert mock_full_backup.location == ".cloud_backup"
+
+    resp = await api_client.post(f"/backups/{mock_full_backup.slug}/restore/full")
+    assert resp.status == 403
+
+    resp = await api_client.post(
+        f"/backups/{mock_full_backup.slug}/restore/partial",
+        json={"homeassistant": True},
+    )
+    assert resp.status == 403
+
+    resp = await api_client.delete(f"/backups/{mock_full_backup.slug}")
+    assert resp.status == 403

tests/api/test_mounts.py

@@ -81,6 +81,7 @@ async def test_api_create_mount(
             "share": "backups",
             "state": "active",
             "read_only": False,
+            "user_path": "/backup/backup_test",
         }
     ]
     coresys.mounts.save_data.assert_called_once()
@@ -257,6 +258,7 @@ async def test_api_update_mount(
             "share": "new_backups",
             "state": "active",
             "read_only": False,
+            "user_path": "/backup/backup_test",
         }
     ]
     coresys.mounts.save_data.assert_called_once()
@@ -292,8 +294,9 @@ async def test_api_update_dbus_error_mount_remains(
     """Test mount remains in list with unsuccessful state if dbus error occurs during update."""
     systemd_service: SystemdService = all_dbus_services["systemd"]
     systemd_unit_service: SystemdUnitService = all_dbus_services["systemd_unit"]
-    systemd_unit_service.active_state = ["failed", "inactive"]
+    systemd_unit_service.active_state = ["failed", "inactive", "failed", "inactive"]
     systemd_service.response_get_unit = [
+        "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
         "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount",
         DBusError("org.freedesktop.systemd1.NoSuchUnit", "error"),
     ]
@@ -325,6 +328,7 @@ async def test_api_update_dbus_error_mount_remains(
             "share": "backups",
             "state": None,
             "read_only": False,
+            "user_path": "/backup/backup_test",
         }
     ]
 
@@ -372,6 +376,7 @@ async def test_api_update_dbus_error_mount_remains(
             "share": "backups",
             "state": None,
             "read_only": False,
+            "user_path": "/backup/backup_test",
         }
     ]
 
@@ -828,6 +833,7 @@ async def test_api_create_read_only_cifs_mount(
             "share": "media",
             "state": "active",
             "read_only": True,
+            "user_path": "/media/media_test",
         }
     ]
     coresys.mounts.save_data.assert_called_once()
@@ -868,6 +874,7 @@ async def test_api_create_read_only_nfs_mount(
             "path": "/media/camera",
             "state": "active",
             "read_only": True,
+            "user_path": "/media/media_test",
         }
     ]
     coresys.mounts.save_data.assert_called_once()