Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-12-24 20:35:55 +00:00
Code Activity

Storage space usage API (#6046)

Browse Source 
* Storage space usage API

* Move to host API

* add tests

* fix test url

* more tests

* fix tests

* fix test

* PR comments

* update test

* tweak format and url

* add .DS_Store to .gitignore

* update tests

* test coverage

* update to new struct

* update test
This commit is contained in:
Petar Petrov
2025-08-19 11:54:53 +03:00
committed by GitHub
parent 43f20fe24f
commit 2324b70084
7 changed files with 746 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

427
tests/api/test_host.py
View File

@@ -381,6 +381,433 @@ 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_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.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
"children": [
{"id": "addon1", "label": "addon1", "used_bytes": 50000000}
],
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 200000000,
"children": [
{"id": "media1", "label": "media1", "used_bytes": 100000000}
],
},
{
"id": "media",
"label": "Media",
"used_bytes": 50000000,
"children": [
{"id": "share1", "label": "share1", "used_bytes": 25000000}
],
},
{
"id": "share",
"label": "Share",
"used_bytes": 300000000,
"children": [
{"id": "backup1", "label": "backup1", "used_bytes": 150000000}
],
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 10000000,
"children": [{"id": "ssl1", "label": "ssl1", "used_bytes": 5000000}],
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 40000000,
"children": [
{
"id": "homeassistant1",
"label": "homeassistant1",
"used_bytes": 20000000,
}
],
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 40000000,
"children": [
{
"id": "homeassistant1",
"label": "homeassistant1",
"used_bytes": 20000000,
}
],
},
]
# Test default max_depth=1
resp = await api_client.get("/host/disks/default/usage")
assert resp.status == 200
result = await resp.json()
assert result["data"]["id"] == "root"
assert result["data"]["label"] == "Root"
assert result["data"]["total_bytes"] == 1000000000
assert result["data"]["used_bytes"] == 500000000
assert "children" in result["data"]
children = result["data"]["children"]
# First child should be system
assert children[0]["id"] == "system"
assert children[0]["label"] == "System"
# Verify all expected directories are present in the remaining children
assert children[1]["id"] == "addons_data"
assert children[2]["id"] == "addons_config"
assert children[3]["id"] == "media"
assert children[4]["id"] == "share"
assert children[5]["id"] == "backup"
assert children[6]["id"] == "ssl"
assert children[7]["id"] == "homeassistant"
# Verify the sizes are correct
assert children[1]["used_bytes"] == 100000000
assert children[2]["used_bytes"] == 200000000
assert children[3]["used_bytes"] == 50000000
assert children[4]["used_bytes"] == 300000000
assert children[5]["used_bytes"] == 10000000
assert children[6]["used_bytes"] == 40000000
assert children[7]["used_bytes"] == 40000000
# Verify system space calculation (total used - sum of known paths)
total_known_space = (
100000000
+ 200000000
+ 50000000
+ 300000000
+ 10000000
+ 40000000
+ 40000000
)
expected_system_space = 500000000 - total_known_space
assert children[0]["used_bytes"] == expected_system_space
# Verify disk_usage was called with supervisor path
mock_disk_usage.assert_called_once_with(coresys.config.path_supervisor)
# Verify get_dir_sizes was called once with all paths
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[0][1] == 1 # max_depth parameter
paths_dict = call_args[0][0] # paths dictionary
assert paths_dict["addons_data"] == coresys.config.path_addons_data
assert paths_dict["addons_config"] == coresys.config.path_addon_configs
assert paths_dict["media"] == coresys.config.path_media
assert paths_dict["share"] == coresys.config.path_share
assert paths_dict["backup"] == coresys.config.path_backup
assert paths_dict["ssl"] == coresys.config.path_ssl
assert paths_dict["homeassistant"] == coresys.config.path_homeassistant
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_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
# Mock deeper directory structure
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "media",
"label": "Media",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "share",
"label": "Share",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
]
# Test with custom max_depth=2
resp = await api_client.get("/host/disks/default/usage?max_depth=2")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
assert result["data"]["children"]
# Verify max_depth=2 was passed to get_dir_sizes
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[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_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 100000000,
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 100000000,
},
{
"id": "media",
"label": "Media",
"used_bytes": 100000000,
},
{
"id": "share",
"label": "Share",
"used_bytes": 100000000,
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 100000000,
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 100000000,
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 100000000,
},
]
# Test with invalid max_depth (non-integer)
resp = await api_client.get("/host/disks/default/usage?max_depth=invalid")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
assert result["data"]["children"]
# Should default to max_depth=1 when invalid value is provided
assert mock_dir_sizes.call_count == 1
call_args = mock_dir_sizes.call_args
assert call_args[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_sizes") as mock_dir_sizes,
):
mock_disk_usage.return_value = (1000000000, 500000000, 500000000)
# Mock empty directory structures (no children)
mock_dir_sizes.return_value = [
{
"id": "addons_data",
"label": "Addons Data",
"used_bytes": 0,
},
{
"id": "addons_config",
"label": "Addons Config",
"used_bytes": 0,
},
{
"id": "media",
"label": "Media",
"used_bytes": 0,
},
{
"id": "share",
"label": "Share",
"used_bytes": 0,
},
{
"id": "backup",
"label": "Backup",
"used_bytes": 0,
},
{
"id": "ssl",
"label": "SSL",
"used_bytes": 0,
},
{
"id": "homeassistant",
"label": "Home Assistant",
"used_bytes": 0,
},
]
resp = await api_client.get("/host/disks/default/usage")
assert resp.status == 200
result = await resp.json()
assert result["data"]["used_bytes"] == 500000000
children = result["data"]["children"]
# First child should be system with all the space
assert children[0]["id"] == "system"
assert children[0]["used_bytes"] == 500000000
# All other directories should have size 0
for i in range(1, len(children)):
assert children[i]["used_bytes"] == 0
@pytest.mark.parametrize("action", ["reboot", "shutdown"])
async def test_migration_blocks_shutdown(
api_client: TestClient,

185
tests/hardware/test_disk.py
View File

@@ -1,6 +1,8 @@
"""Test hardware utils."""
# pylint: disable=protected-access
import errno
import os
from pathlib import Path
from unittest.mock import patch
@@ -9,6 +11,7 @@ import pytest
from supervisor.coresys import CoreSys
from supervisor.hardware.data import Device
from supervisor.resolution.const import UnhealthyReason
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.base import DBusServiceMock
@@ -150,6 +153,141 @@ def test_get_mount_source(coresys):
assert mount_source == "proc"
def test_get_dir_structure_sizes(coresys, tmp_path):
"""Test directory structure size calculation."""
# Create a test directory structure
test_dir = tmp_path / "test_dir"
test_dir.mkdir()
# Create some files
(test_dir / "file1.txt").write_text("content1")
(test_dir / "file2.txt").write_text("content2" * 100) # Larger file
# Create subdirectories
subdir1 = test_dir / "subdir1"
subdir1.mkdir()
(subdir1 / "file3.txt").write_text("content3")
subdir2 = test_dir / "subdir2"
subdir2.mkdir()
(subdir2 / "file4.txt").write_text("content4")
# Create nested subdirectory
nested_dir = subdir1 / "nested"
nested_dir.mkdir()
(nested_dir / "file5.txt").write_text("content5")
# Create a symlink (should be skipped)
(test_dir / "symlink.txt").symlink_to(test_dir / "file1.txt")
# Test with max_depth=1 (default)
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir, max_depth=1)
# Verify the structure
assert result["used_bytes"] > 0
assert "children" not in result
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir, max_depth=2)
# Verify the structure
assert result["used_bytes"] > 0
assert "children" in result
children = result["children"]
# Should have subdir1 and subdir2, but not nested (due to max_depth=1)
child_names = [child["id"] for child in children]
assert "subdir1" in child_names
assert "subdir2" in child_names
assert "nested" not in child_names
# Verify sizes are calculated correctly
subdir1 = next(child for child in children if child["id"] == "subdir1")
subdir2 = next(child for child in children if child["id"] == "subdir2")
assert subdir1["used_bytes"] > 0
assert subdir2["used_bytes"] > 0
assert "children" not in subdir1 # No children due to max_depth=1
assert "children" not in subdir2
# Test with max_depth=2
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir, max_depth=3)
# Should now include nested directory
child_names = [child["id"] for child in result["children"]]
assert "subdir1" in child_names
assert "subdir2" in child_names
subdir1 = next(child for child in result["children"] if child["id"] == "subdir1")
nested_children = [child["id"] for child in subdir1["children"]]
assert "nested" in nested_children
nested = next(child for child in subdir1["children"] if child["id"] == "nested")
assert nested["used_bytes"] > 0
# Test with max_depth=0 (should only count files in root, no children)
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir, max_depth=0)
assert result["used_bytes"] > 0
assert "children" not in result # No children due to max_depth=0
def test_get_dir_structure_sizes_empty_dir(coresys, tmp_path):
"""Test directory structure size calculation with empty directory."""
empty_dir = tmp_path / "empty_dir"
empty_dir.mkdir()
result = coresys.hardware.disk.get_dir_structure_sizes(empty_dir)
assert result["used_bytes"] == 0
assert "children" not in result
def test_get_dir_structure_sizes_nonexistent_dir(coresys, tmp_path):
"""Test directory structure size calculation with nonexistent directory."""
nonexistent_dir = tmp_path / "nonexistent"
result = coresys.hardware.disk.get_dir_structure_sizes(nonexistent_dir)
assert result["used_bytes"] == 0
assert "children" not in result
def test_get_dir_structure_sizes_only_files(coresys, tmp_path):
"""Test directory structure size calculation with only files (no subdirectories)."""
files_dir = tmp_path / "files_dir"
files_dir.mkdir()
# Create some files
(files_dir / "file1.txt").write_text("content1")
(files_dir / "file2.txt").write_text("content2" * 50)
result = coresys.hardware.disk.get_dir_structure_sizes(files_dir)
assert result["used_bytes"] > 0
assert "children" not in result # No children since no subdirectories
def test_get_dir_structure_sizes_zero_size_children(coresys, tmp_path):
"""Test directory structure size calculation with zero-size children."""
test_dir = tmp_path / "zero_size_test"
test_dir.mkdir()
# Create a file in root
(test_dir / "file1.txt").write_text("content1")
# Create an empty subdirectory
empty_subdir = test_dir / "empty_subdir"
empty_subdir.mkdir()
# Create a subdirectory with content
content_subdir = test_dir / "content_subdir"
content_subdir.mkdir()
(content_subdir / "file2.txt").write_text("content2")
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir)
# Should include content_subdir but not empty_subdir (since size > 0)
assert result["used_bytes"] > 0
assert "children" not in result
def test_try_get_emmc_life_time(coresys, tmp_path):
"""Test eMMC life time helper."""
fake_life_time = tmp_path / "fake-mmcblk0-lifetime"
@@ -163,6 +301,53 @@ def test_try_get_emmc_life_time(coresys, tmp_path):
assert value == 10.0
def test_get_dir_structure_sizes_ebadmsg_error(coresys, tmp_path):
"""Test directory structure size calculation with EBADMSG error."""
# Create a test directory structure
test_dir = tmp_path / "test_dir"
test_dir.mkdir()
# Create some files
(test_dir / "file1.txt").write_text("content1")
# Create a subdirectory
subdir = test_dir / "subdir"
subdir.mkdir()
(subdir / "file2.txt").write_text("content2")
# Mock is_dir, is_symlink, and stat methods to handle the EBADMSG error correctly
def mock_is_dir(self):
# Use the real is_dir for all paths
return os.path.isdir(self)
def mock_is_symlink(self):
# Use the real is_symlink for all paths
return os.path.islink(self)
def mock_stat_ebadmsg(self, follow_symlinks=True):
if self == subdir:
raise OSError(errno.EBADMSG, "Bad message")
# For other paths, use the real os.stat
return os.stat(self, follow_symlinks=follow_symlinks)
with (
patch.object(Path, "is_dir", mock_is_dir),
patch.object(Path, "is_symlink", mock_is_symlink),
patch.object(Path, "stat", mock_stat_ebadmsg),
):
result = coresys.hardware.disk.get_dir_structure_sizes(test_dir)
# The EBADMSG error should cause the loop to break, so we get 0 used space
# because the error happens before processing the file in the root directory
assert result["used_bytes"] == 0
assert "children" not in result
# Verify that the unhealthy reason was added
assert coresys.resolution.unhealthy
assert UnhealthyReason.OSERROR_BAD_MESSAGE in coresys.resolution.unhealthy
async def test_try_get_nvme_life_time(
coresys: CoreSys, nvme_data_disk: NVMeControllerService
):