Storage space usage API

supervisor/api/__init__.py

@@ -198,6 +198,7 @@ def _register_host(self) -> None:
                 web.post("/host/reload", api_host.reload),
                 web.post("/host/options", api_host.options),
                 web.get("/host/services", api_host.services),
+                web.get("/host/disk_usage", api_host.disk_usage),
             ]
         )
 

supervisor/api/const.py

@@ -49,6 +49,7 @@
 ATTR_LOCAL_ONLY = "local_only"
 ATTR_LOCATION_ATTRIBUTES = "location_attributes"
 ATTR_LOCATIONS = "locations"
+ATTR_MAX_DEPTH = "max_depth"
 ATTR_MDNS = "mdns"
 ATTR_MODEL = "model"
 ATTR_MOUNTS = "mounts"

supervisor/api/host.py

@@ -51,6 +51,7 @@
     ATTR_FORCE,
     ATTR_IDENTIFIERS,
     ATTR_LLMNR_HOSTNAME,
+    ATTR_MAX_DEPTH,
     ATTR_STARTUP_TIME,
     ATTR_USE_NTP,
     ATTR_VIRTUALIZATION,
@@ -289,3 +290,36 @@ async def advanced_logs(
     ) -> web.StreamResponse:
         """Return systemd-journald logs. Wrapped as standard API handler."""
         return await self.advanced_logs_handler(request, identifier, follow)
+
+    @api_process
+    async def disk_usage(self, request: web.Request) -> dict:
+        """Return a breakdown of storage usage for the system."""
+
+        max_depth = request.query.get(ATTR_MAX_DEPTH, 1)
+        try:
+            max_depth = int(max_depth)
+        except ValueError:
+            max_depth = 1
+
+        disk = self.sys_hardware.disk
+
+        _, used, _ = await self.sys_run_in_executor(
+            disk.disk_usage, self.sys_config.path_supervisor
+        )
+
+        async def dir_info(path):
+            return await self.sys_run_in_executor(
+                disk.get_dir_structure_sizes, path, max_depth
+            )
+
+        return {
+            "size": used,
+            "children": {
+                "addons": await dir_info(self.sys_config.path_addons_data),
+                "media": await dir_info(self.sys_config.path_media),
+                "share": await dir_info(self.sys_config.path_share),
+                "backup": await dir_info(self.sys_config.path_backup),
+                "tmp": await dir_info(self.sys_config.path_tmp),
+                "config": await dir_info(self.sys_config.path_homeassistant),
+            },
+        }

supervisor/hardware/disk.py

@@ -3,6 +3,7 @@
 import logging
 from pathlib import Path
 import shutil
+from typing import Any
 
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import HardwareNotFound
@@ -53,17 +54,65 @@ def get_disk_total_space(self, path: str | Path) -> float:
 
         Must be run in executor.
         """
-        total, _, _ = shutil.disk_usage(path)
+        total, _, _ = self.disk_usage(path)
         return round(total / (1024.0**3), 1)
 
     def get_disk_used_space(self, path: str | Path) -> float:
         """Return used space (GiB) on disk for path.
 
         Must be run in executor.
         """
-        _, used, _ = shutil.disk_usage(path)
+        _, used, _ = self.disk_usage(path)
         return round(used / (1024.0**3), 1)
 
+    def disk_usage(self, path: str | Path) -> tuple[int, int, int]:
+        """Return (total, used, free) in bytes for path.
+
+        Must be run in executor.
+        """
+        return shutil.disk_usage(path)
+
+    def get_dir_structure_sizes(self, path: Path, max_depth: int = 1) -> dict[str, Any]:
+        """Return a recursive dict of subdirectories and their sizes, only if size > 0.
+
+        Excludes external mounts and symlinks to avoid counting files on other filesystems
+        or following symlinks that could lead to infinite loops or incorrect sizes.
+        """
+
+        size = 0
+        if not path.exists():
+            return {"size": size}
+
+        children: dict[str, Any] = {}
+        root_device = path.stat().st_dev
+
+        for child in path.iterdir():
+            if not child.is_dir():
+                size += child.stat(follow_symlinks=False).st_size
+                continue
+
+            # Skip symlinks to avoid infinite loops
+            if child.is_symlink():
+                continue
+
+            try:
+                # Skip if not on same device (external mount)
+                if child.stat().st_dev != root_device:
+                    continue
+            except (OSError, FileNotFoundError):
+                continue
+
+            child_result = self.get_dir_structure_sizes(child, max_depth - 1)
+            if child_result["size"] > 0:
+                size += child_result["size"]
+                if max_depth > 1:
+                    children[child.name] = child_result
+
+        if children:
+            return {"size": size, "children": children}
+
+        return {"size": size}
+
     def get_disk_free_space(self, path: str | Path) -> float:
         """Return free space (GiB) on disk for path.
 

tests/api/test_host.py

@@ -377,6 +377,160 @@ async def test_advanced_logs_errors(coresys: CoreSys, api_client: TestClient):
     )
 
 
+async def test_disk_usage_api(api_client: TestClient, coresys: CoreSys):
+    """Test disk usage API endpoint."""
+    # Mock the disk usage methods
+    with (
+        patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
+        patch.object(
+            coresys.hardware.disk, "get_dir_structure_sizes"
+        ) as mock_dir_sizes,
+    ):
+        # Mock the main disk usage call
+        mock_disk_usage.return_value = (
+            1000000000,
+            500000000,
+            500000000,
+        )  # 1GB total, 500MB used, 500MB free
+
+        # Mock the directory structure sizes for each path
+        mock_dir_sizes.side_effect = [
+            {"size": 100000000, "children": {"addon1": {"size": 50000000}}},  # addons
+            {"size": 200000000, "children": {"media1": {"size": 100000000}}},  # media
+            {"size": 50000000, "children": {"share1": {"size": 25000000}}},  # share
+            {"size": 300000000, "children": {"backup1": {"size": 150000000}}},  # backup
+            {"size": 10000000, "children": {"tmp1": {"size": 5000000}}},  # tmp
+            {"size": 40000000, "children": {"config1": {"size": 20000000}}},  # config
+        ]
+
+        # Test default max_depth=1
+        resp = await api_client.get("/host/disk_usage")
+        assert resp.status == 200
+        result = await resp.json()
+
+        assert result["data"]["size"] == 500000000
+        assert "children" in result["data"]
+        children = result["data"]["children"]
+
+        # Verify all expected directories are present
+        assert "addons" in children
+        assert "media" in children
+        assert "share" in children
+        assert "backup" in children
+        assert "tmp" in children
+        assert "config" in children
+
+        # Verify the sizes are correct
+        assert children["addons"]["size"] == 100000000
+        assert children["media"]["size"] == 200000000
+        assert children["share"]["size"] == 50000000
+        assert children["backup"]["size"] == 300000000
+        assert children["tmp"]["size"] == 10000000
+        assert children["config"]["size"] == 40000000
+
+        # Verify disk_usage was called with supervisor path
+        mock_disk_usage.assert_called_once_with(coresys.config.path_supervisor)
+
+        # Verify get_dir_structure_sizes was called for each directory
+        assert mock_dir_sizes.call_count == 6
+        mock_dir_sizes.assert_any_call(coresys.config.path_addons_data, 1)
+        mock_dir_sizes.assert_any_call(coresys.config.path_media, 1)
+        mock_dir_sizes.assert_any_call(coresys.config.path_share, 1)
+        mock_dir_sizes.assert_any_call(coresys.config.path_backup, 1)
+        mock_dir_sizes.assert_any_call(coresys.config.path_tmp, 1)
+        mock_dir_sizes.assert_any_call(coresys.config.path_homeassistant, 1)
+
+
+async def test_disk_usage_api_with_custom_depth(
+    api_client: TestClient, coresys: CoreSys
+):
+    """Test disk usage API endpoint with custom max_depth parameter."""
+    with (
+        patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
+        patch.object(
+            coresys.hardware.disk, "get_dir_structure_sizes"
+        ) as mock_dir_sizes,
+    ):
+        mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
+
+        # Mock deeper directory structure
+        mock_dir_sizes.side_effect = [
+            {
+                "size": 100000000,
+                "children": {
+                    "addon1": {
+                        "size": 50000000,
+                        "children": {"subdir1": {"size": 25000000}},
+                    }
+                },
+            }
+        ] * 6  # Same structure for all directories
+
+        # Test with custom max_depth=2
+        resp = await api_client.get("/host/disk_usage?max_depth=2")
+        assert resp.status == 200
+        result = await resp.json()
+        assert result["data"]["size"] == 500000000
+        assert result["data"]["children"]
+
+        # Verify max_depth=2 was passed to get_dir_structure_sizes
+        assert mock_dir_sizes.call_count == 6
+        for call in mock_dir_sizes.call_args_list:
+            assert call[0][1] == 2  # max_depth parameter
+
+
+async def test_disk_usage_api_invalid_depth(api_client: TestClient, coresys: CoreSys):
+    """Test disk usage API endpoint with invalid max_depth parameter."""
+    with (
+        patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
+        patch.object(
+            coresys.hardware.disk, "get_dir_structure_sizes"
+        ) as mock_dir_sizes,
+    ):
+        mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
+        mock_dir_sizes.return_value = {"size": 100000000}
+
+        # Test with invalid max_depth (non-integer)
+        resp = await api_client.get("/host/disk_usage?max_depth=invalid")
+        assert resp.status == 200
+        result = await resp.json()
+        assert result["data"]["size"] == 500000000
+        assert result["data"]["children"]
+
+        # Should default to max_depth=1 when invalid value is provided
+        assert mock_dir_sizes.call_count == 6
+        for call in mock_dir_sizes.call_args_list:
+            assert call[0][1] == 1  # Should default to 1
+
+
+async def test_disk_usage_api_empty_directories(
+    api_client: TestClient, coresys: CoreSys
+):
+    """Test disk usage API endpoint with empty directories."""
+    with (
+        patch.object(coresys.hardware.disk, "disk_usage") as mock_disk_usage,
+        patch.object(
+            coresys.hardware.disk, "get_dir_structure_sizes"
+        ) as mock_dir_sizes,
+    ):
+        mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
+
+        # Mock empty directory structures (no children)
+        mock_dir_sizes.return_value = {"size": 0}
+
+        resp = await api_client.get("/host/disk_usage")
+        assert resp.status == 200
+        result = await resp.json()
+
+        assert result["data"]["size"] == 500000000
+        children = result["data"]["children"]
+
+        # All directories should have size 0
+        for directory in children.values():
+            assert directory["size"] == 0
+            assert "children" not in directory  # No children when size is 0
+
+
 @pytest.mark.parametrize("action", ["reboot", "shutdown"])
 async def test_migration_blocks_shutdown(
     api_client: TestClient,