Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2026-07-04 04:15:07 +01:00
Code Activity
Files
c6c8e5a65e2ac1cd044ac312b4e4f019ff9dd896
supervisor/tests/api/test_host.py
T
Stefan Agner 81e235376e Fix typos repo-wide and add codespell pre-commit hook (#6949) 
* Fix typos in comments, docstrings and log messages

Correct 39 spelling mistakes across comments, docstrings and log/error
message strings throughout the package (e.g. "conection" -> "connection",
"Incomming" -> "Incoming", "Rasie" -> "Raise"). All changes are confined
to human-readable text; no identifiers, attributes or D-Bus contracts are
touched, so there is no behavior change. Found with codespell.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Fix typos in tests and CI workflow

Correct spelling mistakes in test comments, docstrings and data, plus one
in the builder workflow, so the whole tree is clean for the codespell hook
added next. The assertion in test_network_manager.py is updated to match
the corrected "Unknown error while processing" log message in the source.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Add codespell pre-commit hook

Wire up codespell so spelling mistakes in comments, docstrings and strings
are caught automatically. The vendored frontend panel is excluded, and
"hass" and "astroid" are added to the ignore list as known false positives
(the Home Assistant abbreviation and the pylint dependency package).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Address review feedback

Improve grammar in several of the touched comments and docstrings: use the
plural "ignore conditions" for the list-returning property, add the missing
auxiliary verb and fix agreement in the timezone-filter comment, fix
"backups ... use" agreement, and reword "underlay" to "underlying" in the
arch module docstring.

Also drop the "*.json" skip from the codespell hook. It was carried over
from another project but is unnecessary here (all tracked JSON is clean),
and skipping it would needlessly leave translation and data JSON unchecked.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Reword onboarding comment

"overflight" was a literal calque of the German "überflogen"; use the
idiomatic "skimmed through" instead.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-16 14:09:16 +02:00

937 lines
32 KiB
Python
Raw Blame History

"""Test Host API."""
from collections.abc import AsyncGenerator
from datetime import UTC, datetime
from unittest.mock import ANY, MagicMock, patch
from aiohttp.test_utils import TestClient
from dbus_fast import DBusError, ErrorType
import pytest
import time_machine
from supervisor.coresys import CoreSys
from supervisor.dbus.resolved import Resolved
from supervisor.homeassistant.api import APIState
from supervisor.host.const import LogFormat, LogFormatter
from supervisor.host.control import SystemControl
from tests.dbus_service_mocks.base import DBusServiceMock
from tests.dbus_service_mocks.hostname import Hostname as HostnameService
from tests.dbus_service_mocks.systemd import Systemd as SystemdService
DEFAULT_RANGE = "entries=:-99:100"
DEFAULT_RANGE_FOLLOW = "entries=:-99:18446744073709551615"
# pylint: disable=protected-access
@pytest.fixture(name="coresys_disk_info")
async def fixture_coresys_disk_info(coresys: CoreSys) -> AsyncGenerator[CoreSys]:
"""Mock basic disk information for host APIs."""
async def mock_disk_lifetime(_):
return 0
coresys.hardware.disk.get_disk_life_time = mock_disk_lifetime
coresys.hardware.disk.get_disk_free_space = lambda _: 5000
coresys.hardware.disk.get_disk_total_space = lambda _: 50000
coresys.hardware.disk.get_disk_used_space = lambda _: 45000
return coresys
async def test_api_host_info(
api_client_with_prefix: tuple[TestClient, str], coresys_disk_info: CoreSys
):
"""Test host info api."""
api_client, prefix = api_client_with_prefix
coresys = coresys_disk_info
dt_utc = datetime(2026, 2, 17, 1, 23, 45, 678901, tzinfo=UTC)
await coresys.dbus.agent.connect(coresys.dbus.bus)
await coresys.dbus.agent.update()
with time_machine.travel(dt_utc, tick=False):
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert result["data"]["apparmor_version"] == "2.13.2"
assert result["data"]["dt_utc"] == "2026-02-17T01:23:45.678901+00:00"
async def test_api_host_features(
api_client_with_prefix: tuple[TestClient, str],
coresys_disk_info: CoreSys,
dbus_is_connected,
):
"""Test host info features."""
api_client, prefix = api_client_with_prefix
coresys = coresys_disk_info
coresys.host.sys_dbus.systemd.is_connected = False
coresys.host.sys_dbus.network.is_connected = False
coresys.host.sys_dbus.hostname.is_connected = False
coresys.host.sys_dbus.timedate.is_connected = False
coresys.host.sys_dbus.agent.is_connected = False
coresys.host.sys_dbus.resolved.is_connected = False
coresys.host.sys_dbus.udisks2.is_connected = False
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "reboot" not in result["data"]["features"]
assert "services" not in result["data"]["features"]
assert "shutdown" not in result["data"]["features"]
assert "network" not in result["data"]["features"]
assert "hostname" not in result["data"]["features"]
assert "timedate" not in result["data"]["features"]
assert "os_agent" not in result["data"]["features"]
assert "resolved" not in result["data"]["features"]
assert "disk" not in result["data"]["features"]
coresys.host.sys_dbus.systemd.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "reboot" in result["data"]["features"]
assert "services" in result["data"]["features"]
assert "shutdown" in result["data"]["features"]
coresys.host.sys_dbus.network.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "network" in result["data"]["features"]
coresys.host.sys_dbus.hostname.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "hostname" in result["data"]["features"]
coresys.host.sys_dbus.timedate.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "timedate" in result["data"]["features"]
coresys.host.sys_dbus.agent.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "os_agent" in result["data"]["features"]
coresys.host.sys_dbus.resolved.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "resolved" in result["data"]["features"]
coresys.host.sys_dbus.udisks2.is_connected = True
coresys.host.supported_features.cache_clear()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert "disk" in result["data"]["features"]
async def test_api_llmnr_mdns_info(
api_client_with_prefix: tuple[TestClient, str], coresys_disk_info: CoreSys
):
"""Test llmnr and mdns details in info."""
api_client, prefix = api_client_with_prefix
coresys = coresys_disk_info
# pylint: disable=protected-access
coresys.host.sys_dbus._resolved = Resolved()
# pylint: enable=protected-access
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert result["data"]["broadcast_llmnr"] is None
assert result["data"]["broadcast_mdns"] is None
assert result["data"]["llmnr_hostname"] is None
await coresys.dbus.resolved.connect(coresys.dbus.bus)
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert result["data"]["broadcast_llmnr"] is True
assert result["data"]["broadcast_mdns"] is False
assert result["data"]["llmnr_hostname"] == "homeassistant"
async def test_api_boot_ids_info(
api_client_with_prefix: tuple[TestClient, str], journald_logs: MagicMock
):
"""Test getting boot IDs."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/host/logs/boots")
result = await resp.json()
assert result["data"] == {"boots": {"0": "ccc", "-1": "bbb", "-2": "aaa"}}
async def test_api_identifiers_info(
api_client_with_prefix: tuple[TestClient, str], journald_logs: MagicMock
):
"""Test getting syslog identifiers."""
api_client, prefix = api_client_with_prefix
resp = await api_client.get(f"{prefix}/host/logs/identifiers")
result = await resp.json()
assert result["data"] == {
"identifiers": ["hassio_supervisor", "hassos-config", "kernel"]
}
async def test_api_virtualization_info(
api_client_with_prefix: tuple[TestClient, str],
all_dbus_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
coresys_disk_info: CoreSys,
):
"""Test getting virtualization info."""
api_client, prefix = api_client_with_prefix
systemd_service: SystemdService = all_dbus_services["systemd"]
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert result["data"]["virtualization"] == ""
systemd_service.virtualization = "vmware"
await coresys_disk_info.dbus.systemd.update()
resp = await api_client.get(f"{prefix}/host/info")
result = await resp.json()
assert result["data"]["virtualization"] == "vmware"
async def test_advanced_logs(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
journald_logs: MagicMock,
):
"""Test advanced logging API entries with identifier and custom boot."""
api_client, prefix = api_client_with_prefix
await api_client.get(f"{prefix}/host/logs")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
identifier = "dropbear"
await api_client.get(f"{prefix}/host/logs/identifiers/{identifier}")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": identifier},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
bootid = "798cc03bcd77465482b6a1c43dc6a5fc"
await api_client.get(f"{prefix}/host/logs/boots/{bootid}")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": bootid,
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get(f"{prefix}/host/logs/boots/{bootid}/identifiers/{identifier}")
journald_logs.assert_called_once_with(
params={"_BOOT_ID": bootid, "SYSLOG_IDENTIFIER": identifier},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
headers = {"Range": "entries=:-19:10"}
await api_client.get(f"{prefix}/host/logs", headers=headers)
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=headers["Range"],
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get(f"{prefix}/host/logs/follow")
journald_logs.assert_called_once_with(
params={
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
"follow": "",
},
range_header=DEFAULT_RANGE_FOLLOW,
accept=LogFormat.JOURNAL,
)
# Host logs don't have a /latest endpoint
resp = await api_client.get(f"{prefix}/host/logs/latest")
assert resp.status == 404
async def test_advanced_logs_query_parameters(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
journald_logs: MagicMock,
journal_logs_reader: MagicMock,
):
"""Test advanced logging API entries controlled by query parameters."""
api_client, prefix = api_client_with_prefix
# Check lines query parameter
await api_client.get(f"{prefix}/host/logs?lines=53")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header="entries=:-52:53",
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
# Check verbose logs formatter via query parameter
await api_client.get(f"{prefix}/host/logs?verbose")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journal_logs_reader.assert_called_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
journald_logs.reset_mock()
# Query parameters should take precedence over headers
await api_client.get(
f"{prefix}/host/logs?lines=53&verbose",
headers={
"Range": "entries=:-19:10",
"Accept": "text/plain",
},
)
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header="entries=:-52:53",
accept=LogFormat.JOURNAL,
)
journal_logs_reader.assert_called_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
journald_logs.reset_mock()
# Check no_colors query parameter
await api_client.get(f"{prefix}/host/logs?no_colors")
journald_logs.assert_called_once_with(
params={"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journal_logs_reader.assert_called_with(ANY, LogFormatter.VERBOSE, True)
async def test_advanced_logs_boot_id_offset(
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
journald_logs: MagicMock,
):
"""Test advanced logging API when using an offset as boot ID."""
api_client, prefix = api_client_with_prefix
await api_client.get(f"{prefix}/host/logs/boots/0")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "ccc",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get(f"{prefix}/host/logs/boots/-2")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "aaa",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
await api_client.get(f"{prefix}/host/logs/boots/2")
journald_logs.assert_called_once_with(
params={
"_BOOT_ID": "bbb",
"SYSLOG_IDENTIFIER": coresys.host.logs.default_identifiers,
},
range_header=DEFAULT_RANGE,
accept=LogFormat.JOURNAL,
)
journald_logs.reset_mock()
async def test_advanced_logs_formatters(
journald_gateway: MagicMock,
api_client_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
journal_logs_reader: MagicMock,
):
"""Test advanced logs formatters varying on Accept header."""
api_client, prefix = api_client_with_prefix
await api_client.get(f"{prefix}/host/logs")
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
headers = {"Accept": "text/x-log"}
await api_client.get(f"{prefix}/host/logs", headers=headers)
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
await api_client.get(f"{prefix}/host/logs/identifiers/test")
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.PLAIN, False)
journal_logs_reader.reset_mock()
headers = {"Accept": "text/x-log"}
await api_client.get(f"{prefix}/host/logs/identifiers/test", headers=headers)
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.VERBOSE, False)
journal_logs_reader.reset_mock()
await api_client.get(
f"{prefix}/host/logs/identifiers/test", skip_auto_headers={"Accept"}
)
journal_logs_reader.assert_called_once_with(ANY, LogFormatter.PLAIN, False)
async def test_advanced_logs_errors(
coresys: CoreSys, api_client_with_prefix: tuple[TestClient, str]
):
"""Test advanced logging API errors."""
api_client, prefix = api_client_with_prefix
with patch("supervisor.host.logs.SYSTEMD_JOURNAL_GATEWAYD_SOCKET") as socket:
socket.is_socket.return_value = False
await coresys.host.logs.post_init()
resp = await api_client.get(f"{prefix}/host/logs")
assert resp.content_type == "text/plain"
assert resp.status == 400
content = await resp.text()
assert content == "No systemd-journal-gatewayd Unix socket available"
headers = {"Accept": "application/json"}
resp = await api_client.get(f"{prefix}/host/logs", headers=headers)
assert resp.content_type == "text/plain"
assert resp.status == 400
content = await resp.text()
assert (
content
== "Invalid content type requested. Only text/plain and text/x-log supported for now."
)
async def test_disk_usage_api(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test disk usage API endpoint."""
api_client, prefix = api_client_with_prefix
# 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": "Apps Data",
"used_bytes": 100000000,
"children": [
{"id": "addon1", "label": "addon1", "used_bytes": 50000000}
],
},
{
"id": "addons_config",
"label": "Apps 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(f"{prefix}/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_apps_data
assert paths_dict["addons_config"] == coresys.config.path_app_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_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test disk usage API endpoint with custom max_depth parameter."""
api_client, prefix = api_client_with_prefix
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": "Apps Data",
"used_bytes": 100000000,
"children": [
{
"id": "addon1",
"label": "addon1",
"used_bytes": 50000000,
"children": [
{
"id": "subdir1",
"label": "subdir1",
"used_bytes": 25000000,
},
],
},
],
},
{
"id": "addons_config",
"label": "Apps 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(f"{prefix}/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_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test disk usage API endpoint with invalid max_depth parameter."""
api_client, prefix = api_client_with_prefix
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": "Apps Data",
"used_bytes": 100000000,
},
{
"id": "addons_config",
"label": "Apps 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(
f"{prefix}/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_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test disk usage API endpoint with empty directories."""
api_client, prefix = api_client_with_prefix
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": "Apps Data",
"used_bytes": 0,
},
{
"id": "addons_config",
"label": "Apps 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(f"{prefix}/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_with_prefix: tuple[TestClient, str],
coresys: CoreSys,
action: str,
):
"""Test that an offline db migration in progress stops users from shutting down or rebooting system."""
api_client, prefix = api_client_with_prefix
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
resp = await api_client.post(f"{prefix}/host/{action}")
assert resp.status == 503
result = await resp.json()
assert (
result["message"]
== "Home Assistant offline database migration in progress, please wait until complete before shutting down host"
)
async def test_force_reboot_during_migration(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test force option reboots even during a migration."""
api_client, prefix = api_client_with_prefix
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
with patch.object(SystemControl, "reboot") as reboot:
await api_client.post(f"{prefix}/host/reboot", json={"force": True})
reboot.assert_called_once()
async def test_force_shutdown_during_migration(
api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
"""Test force option shutdown even during a migration."""
api_client, prefix = api_client_with_prefix
coresys.homeassistant.api.get_api_state.return_value = APIState("NOT_RUNNING", True)
with patch.object(SystemControl, "shutdown") as shutdown:
await api_client.post(f"{prefix}/host/shutdown", json={"force": True})
shutdown.assert_called_once()
async def test_set_hostname_invalid_returns_400(
api_client: TestClient,
all_dbus_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
):
"""An INVALID_ARGS rejection from hostnamed becomes a 400 with a structured body."""
hostname_service: HostnameService = all_dbus_services["hostname"]
hostname_service.response_set_static_hostname = DBusError(
ErrorType.INVALID_ARGS, "Invalid static hostname 'bad name'"
)
resp = await api_client.post("/host/options", json={"hostname": "bad name"})
assert resp.status == 400
body = await resp.json()
assert body["result"] == "error"
assert body["message"] == "Invalid hostname 'bad name'"
assert body["error_key"] == "host_invalid_hostname"
assert body["extra_fields"] == {"hostname": "bad name"}
View Git Blame Copy Permalink