supervisor/tests/api/test_host.py

"""Test Host API."""

import asyncio
from collections.abc import AsyncGenerator
from contextlib import suppress
from datetime import UTC, datetime
import errno
import gc
import time
from typing import Any
from unittest.mock import ANY, MagicMock, patch

from aiohttp import ClientPayloadError
from aiohttp.test_utils import TestClient
from dbus_fast import DBusError, ErrorType
import pytest
import time_machine

from supervisor.api.host import APIHost
from supervisor.coresys import CoreSys
from supervisor.dbus.const import UnitActiveState
from supervisor.dbus.resolved import Resolved
from supervisor.exceptions import (
    HostJournalGatewaydConnectionError,
    MountUsageTimeoutError,
)
from supervisor.homeassistant.api import APIState
from supervisor.host.const import LogFormat, LogFormatter
from supervisor.host.control import SystemControl
from supervisor.mounts.mount import Mount

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_advanced_logs_gateway_closed_mid_stream(
    journald_gateway: MagicMock,
    api_client_with_prefix: tuple[TestClient, str],
):
    """Test connection to journal gateway closed mid-stream ends the stream gracefully."""
    api_client, prefix = api_client_with_prefix

    journald_gateway.content.feed_data(b"__CURSOR=cursor1\nMESSAGE=Hello, world!\n\n")

    resp = await api_client.get(f"{prefix}/host/logs/identifiers/test")
    assert resp.status == 200

    # Simulate connection to systemd-journal-gatewayd being closed mid-stream,
    # e.g. because it was stopped on host shutdown.
    journald_gateway.content.set_exception(
        ClientPayloadError("Response payload is not completed")
    )

    assert await resp.text() == "Hello, world!\n"


async def test_advanced_logs_gateway_reset_before_stream(
    journald_gateway: MagicMock,
    api_client_with_prefix: tuple[TestClient, str],
):
    """Test connection reset before the log stream started returns an API error."""
    api_client, prefix = api_client_with_prefix

    journald_gateway.content.set_exception(
        ClientPayloadError("Response payload is not completed")
    )

    resp = await api_client.get(f"{prefix}/host/logs/identifiers/test")
    assert resp.status == 400
    assert (
        await resp.text()
        == "Connection reset when trying to fetch data from systemd-journald."
    )


async def test_advanced_logs_gateway_unavailable(
    api_client_with_prefix: tuple[TestClient, str],
    journald_logs: MagicMock,
    caplog: pytest.LogCaptureFixture,
):
    """Test connection failure to journal gateway returns a plain API error."""
    api_client, prefix = api_client_with_prefix

    journald_logs.side_effect = HostJournalGatewaydConnectionError(
        "Unable to connect to systemd-journal-gatewayd"
    )

    with patch("supervisor.api.utils.async_capture_exception") as capture_exception:
        resp = await api_client.get(f"{prefix}/host/logs")

    assert resp.status == 400
    assert await resp.text() == "Unable to connect to systemd-journal-gatewayd"
    capture_exception.assert_not_called()
    assert "Unexpected error during API call" not in caplog.text


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": "apps_data",
                "label": "Apps Data",
                "used_bytes": 100000000,
                "children": [
                    {"id": "addon1", "label": "addon1", "used_bytes": 50000000}
                ],
            },
            {
                "id": "apps_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
        expected_data_id = "addons_data" if prefix == "" else "apps_data"
        expected_config_id = "addons_config" if prefix == "" else "apps_config"
        assert children[1]["id"] == expected_data_id
        assert children[2]["id"] == expected_config_id
        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["apps_data"] == coresys.config.path_apps_data
        assert paths_dict["apps_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": "apps_data",
                "label": "Apps Data",
                "used_bytes": 100000000,
                "children": [
                    {
                        "id": "addon1",
                        "label": "addon1",
                        "used_bytes": 50000000,
                        "children": [
                            {
                                "id": "subdir1",
                                "label": "subdir1",
                                "used_bytes": 25000000,
                            },
                        ],
                    },
                ],
            },
            {
                "id": "apps_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": "apps_data",
                "label": "Apps Data",
                "used_bytes": 100000000,
            },
            {
                "id": "apps_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": "apps_data",
                "label": "Apps Data",
                "used_bytes": 0,
            },
            {
                "id": "apps_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


async def test_disk_usage_api_v1_uses_legacy_addon_ids(
    api_client: TestClient, coresys: CoreSys
):
    """Test v1 disk usage response uses legacy addon IDs."""
    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": "apps_data", "label": "Apps Data", "used_bytes": 100000000},
            {
                "id": "apps_config",
                "label": "Apps Config",
                "used_bytes": 200000000,
            },
            {"id": "media", "label": "Media", "used_bytes": 50000000},
            {"id": "share", "label": "Share", "used_bytes": 300000000},
            {"id": "backup", "label": "Backup", "used_bytes": 10000000},
            {"id": "ssl", "label": "SSL", "used_bytes": 40000000},
            {
                "id": "homeassistant",
                "label": "Home Assistant",
                "used_bytes": 40000000,
            },
        ]

        resp = await api_client.get("/host/disks/default/usage")
        assert resp.status == 200
        result = await resp.json()

        child_ids = [child["id"] for child in result["data"]["children"]]
        assert "addons_data" in child_ids
        assert "addons_config" in child_ids
        assert "apps_data" not in child_ids
        assert "apps_config" not in child_ids


@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"}


def _register_mount(coresys: CoreSys, name: str, state: UnitActiveState) -> Mount:
    """Register a CIFS mount with the manager in the given systemd state."""
    mount = Mount.from_dict(
        coresys,
        {
            "name": name,
            "type": "cifs",
            "usage": "media",
            "server": "media.local",
            "share": "media",
        },
    )
    mount._state = state
    coresys.mounts._mounts = {mount.name: mount}
    return mount


@pytest.fixture(name="active_mount")
async def fixture_active_mount(
    coresys: CoreSys, tmp_supervisor_data, path_extern
) -> Mount:
    """Return an active CIFS mount registered with the mount manager."""
    return _register_mount(coresys, "media_test", UnitActiveState.ACTIVE)


@pytest.mark.parametrize(
    "query",
    ["", "?max_depth=0", "?max_depth=1"],
    ids=["depth-default", "depth-explicit-0", "depth-1-emits-nothing"],
)
async def test_disk_usage_api_mount_totals_only(
    api_client_with_prefix: tuple[TestClient, str],
    coresys: CoreSys,
    active_mount: Mount,
    query: str,
):
    """Test a mount reports totals only without ever walking the tree.

    The directory walker recurses regardless of max_depth and only emits
    children beyond depth 1, so whenever it could not produce output — depth 0
    and depth 1 alike — it has to be skipped outright. Calling it would walk
    the whole mount on every request just to discard the result.
    """
    api_client, prefix = api_client_with_prefix

    with (
        patch.object(coresys.hardware.disk, "disk_usage_for_mount") as mock_disk_usage,
        patch.object(
            coresys.hardware.disk, "get_dir_structure_sizes"
        ) as mock_structure,
    ):
        # Middle value is deliberately wrong: used must come from total - free so
        # that reserved space counts as used, as it does for the system disk.
        mock_disk_usage.return_value = (2000000000, 999, 800000000)

        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage{query}")

    assert resp.status == 200
    result = await resp.json()
    assert result["data"] == {
        "id": "media_test",
        "label": "media_test",
        "total_bytes": 2000000000,
        "used_bytes": 1200000000,
    }
    # Omitted entirely rather than empty, like every other node in the tree
    assert "children" not in result["data"]
    mock_structure.assert_not_called()
    mock_disk_usage.assert_called_once_with(active_mount.local_where)


async def test_disk_usage_api_mount_breakdown(
    api_client_with_prefix: tuple[TestClient, str],
    coresys: CoreSys,
    active_mount: Mount,
):
    """Test a mount breakdown at depth 2.

    Depth means the same thing as it does for the system disk: level 1 is the
    labeled known paths, which a mount does not have, so a mount's own
    subdirectories start at level 2.
    """
    api_client, prefix = api_client_with_prefix

    with (
        patch.object(coresys.hardware.disk, "disk_usage_for_mount") as mock_disk_usage,
        patch.object(
            coresys.hardware.disk, "get_dir_structure_sizes"
        ) as mock_structure,
    ):
        mock_disk_usage.return_value = (2000000000, 999, 800000000)
        mock_structure.return_value = {
            "used_bytes": 1100000000,
            "children": [
                {"id": "movies", "label": "movies", "used_bytes": 900000000},
                {"id": "music", "label": "music", "used_bytes": 200000000},
            ],
        }

        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage?max_depth=2")

    assert resp.status == 200
    result = await resp.json()
    children = result["data"]["children"]

    # What the walk could not attribute is reported as "other", so a mount's
    # children add up to its used_bytes like every other node in the tree.
    assert children == [
        {"id": "movies", "label": "movies", "used_bytes": 900000000},
        {"id": "music", "label": "music", "used_bytes": 200000000},
        {"id": "other", "label": "Other", "used_bytes": 100000000},
    ]
    assert (
        sum(child["used_bytes"] for child in children) == (result["data"]["used_bytes"])
    )
    # check_oserror off: read errors from a mount walk must stay the mount's
    # problem instead of marking the whole system unhealthy
    mock_structure.assert_called_once_with(
        active_mount.local_where, 2, check_oserror=False
    )


@pytest.mark.parametrize(
    "walked_bytes",
    [1200000000, 1300000000],
    ids=["accounts-for-everything", "overshoots-the-filesystem"],
)
@pytest.mark.usefixtures("active_mount")
async def test_disk_usage_api_mount_breakdown_without_remainder(
    api_client_with_prefix: tuple[TestClient, str],
    coresys: CoreSys,
    walked_bytes: int,
):
    """Test no "other" child when there is nothing left to attribute.

    A walk that accounts for everything needs no remainder, and one that
    overshoots (racing a deletion against the filesystem figure) must not
    produce a negative one.
    """
    api_client, prefix = api_client_with_prefix

    with (
        patch.object(coresys.hardware.disk, "disk_usage_for_mount") as mock_disk_usage,
        patch.object(
            coresys.hardware.disk, "get_dir_structure_sizes"
        ) as mock_structure,
    ):
        # used = total - free = 1200000000
        mock_disk_usage.return_value = (2000000000, 999, 800000000)
        mock_structure.return_value = {
            "used_bytes": walked_bytes,
            "children": [
                {"id": "movies", "label": "movies", "used_bytes": walked_bytes},
            ],
        }

        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage?max_depth=2")

    assert resp.status == 200
    result = await resp.json()
    assert result["data"]["children"] == [
        {"id": "movies", "label": "movies", "used_bytes": walked_bytes},
    ]
    assert all(child["id"] != "other" for child in result["data"]["children"])


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_disk_usage_api_unknown_mount(
    api_client_with_prefix: tuple[TestClient, str],
):
    """Test requesting usage for a mount that does not exist."""
    api_client, prefix = api_client_with_prefix

    resp = await api_client.get(f"{prefix}/host/disks/nope/usage")

    assert resp.status == 404
    result = await resp.json()
    assert result["message"] == "No mount exists with name nope"
    assert result["error_key"] == "mount_not_found_error"
    assert result["extra_fields"] == {"name": "nope"}


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_disk_usage_api_inactive_mount(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test a mount that is not active cannot report usage."""
    api_client, prefix = api_client_with_prefix
    _register_mount(coresys, "media_test", UnitActiveState.FAILED)

    with patch.object(coresys.hardware.disk, "disk_usage_for_mount") as mock_disk_usage:
        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage")

    assert resp.status == 400
    result = await resp.json()
    assert result["error_key"] == "mount_usage_not_active_error"
    assert result["extra_fields"] == {"name": "media_test"}
    # Rejected before any probe is attempted
    mock_disk_usage.assert_not_called()


@pytest.mark.usefixtures("active_mount")
async def test_disk_usage_api_mount_timeout(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test a probe that never returns fails cleanly for that mount."""
    api_client, prefix = api_client_with_prefix

    def _never_returns(_):
        time.sleep(0.5)
        return (1, 1, 1)

    with (
        patch("supervisor.api.host.MOUNT_USAGE_TIMEOUT", 0.05),
        patch.object(
            coresys.hardware.disk, "disk_usage_for_mount", side_effect=_never_returns
        ),
    ):
        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage")

    assert resp.status == 400
    result = await resp.json()
    assert result["error_key"] == "mount_usage_timeout_error"
    assert result["extra_fields"] == {"name": "media_test"}


@pytest.mark.usefixtures("active_mount")
async def test_disk_usage_api_mount_unreachable(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test an unreachable mount reports an error rather than failing the API."""
    api_client, prefix = api_client_with_prefix

    with patch.object(
        coresys.hardware.disk,
        "disk_usage_for_mount",
        side_effect=OSError(errno.EHOSTDOWN, "Host is down"),
    ):
        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage")

    assert resp.status == 400
    result = await resp.json()
    assert result["error_key"] == "mount_usage_read_error"
    assert result["extra_fields"]["name"] == "media_test"
    assert "Host is down" in result["extra_fields"]["reason"]


@pytest.mark.usefixtures("active_mount")
async def test_disk_usage_api_ghost_mount(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test a ghost mount is reported unreadable, not misreported.

    systemd can still report a unit active after the path stopped being a
    mount point; statvfs then silently returns the host data disk's numbers.
    Serving those under the mount's name would be a plausible-looking lie.
    """
    api_client, prefix = api_client_with_prefix

    with (
        patch.object(coresys.hardware.disk, "disk_usage_for_mount", return_value=None),
        patch.object(
            coresys.hardware.disk, "get_dir_structure_sizes"
        ) as mock_structure,
    ):
        resp = await api_client.get(f"{prefix}/host/disks/media_test/usage?max_depth=2")

    assert resp.status == 400
    result = await resp.json()
    assert result["error_key"] == "mount_usage_not_mounted_error"
    assert result["extra_fields"] == {"name": "media_test"}
    # A ghost is never walked either
    mock_structure.assert_not_called()


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_mount_usage_caller_timeout_joins_same_probe(coresys: CoreSys):
    """Test a caller's timeout neither cancels nor orphans the shared probe.

    A probe killed by its own timeout would be dropped from the registry while
    its executor thread stays parked in the kernel, so the next request would
    stack a fresh thread against the same mount. The timeout therefore bounds
    only the caller's wait: the probe and its registry entry outlive slow
    callers, and later callers join the same probe.
    """
    api_host = APIHost()
    api_host.coresys = coresys
    mount = _register_mount(coresys, "media_test", UnitActiveState.ACTIVE)

    def _slow(_):
        time.sleep(0.4)
        return (2000000000, 999, 800000000)

    with patch.object(
        coresys.hardware.disk, "disk_usage_for_mount", side_effect=_slow
    ) as mock_usage:
        with (
            patch("supervisor.api.host.MOUNT_USAGE_TIMEOUT", 0.05),
            pytest.raises(
                MountUsageTimeoutError, match="Timed out reading storage usage"
            ),
        ):
            await api_host._mount_usage(mount, 0)

        # The probe survived its caller: still registered, still running
        assert ("media_test", 0) in api_host._mount_usage_probes

        # A later caller (with the real, generous timeout) joins that same
        # probe rather than starting another executor thread
        result = await api_host._mount_usage(mount, 0)

    assert result["used_bytes"] == 1200000000
    mock_usage.assert_called_once()
    # Completion popped the entry, so the next request starts fresh
    assert not api_host._mount_usage_probes


@pytest.mark.usefixtures("active_mount")
async def test_disk_usage_api_mount_shares_in_flight_probe(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test concurrent requests for the same mount share one probe.

    A probe of an unreachable mount blocks until the kernel gives up, so
    stacking one executor thread per caller is what this guards against.
    """
    api_client, prefix = api_client_with_prefix

    def _slow(_):
        time.sleep(0.3)
        return (2000000000, 999, 800000000)

    with patch.object(
        coresys.hardware.disk, "disk_usage_for_mount", side_effect=_slow
    ) as mock_disk_usage:
        first, second = await asyncio.gather(
            api_client.get(f"{prefix}/host/disks/media_test/usage"),
            api_client.get(f"{prefix}/host/disks/media_test/usage"),
        )

    assert first.status == 200
    assert second.status == 200
    assert (await first.json())["data"] == (await second.json())["data"]
    mock_disk_usage.assert_called_once()


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_mount_usage_abandoned_probe_retrieves_its_exception(coresys: CoreSys):
    """Test an abandoned failing probe is not reported as an unretrieved exception.

    The shield deliberately keeps a probe alive past the caller that started it,
    so a browser giving up on an unreachable mount leaves a probe that still
    fails later with nobody waiting on it. Left unretrieved, asyncio reports it
    with a traceback that reads like a supervisor fault rather than a mount that
    was never going to answer.
    """
    api_host = APIHost()
    api_host.coresys = coresys
    mount = _register_mount(coresys, "media_test", UnitActiveState.ACTIVE)

    reported: list[dict[str, Any]] = []
    loop = asyncio.get_running_loop()
    previous_handler = loop.get_exception_handler()
    loop.set_exception_handler(lambda _loop, context: reported.append(context))

    def _fails_slowly(_):
        time.sleep(0.2)
        raise OSError(errno.EHOSTDOWN, "Host is down")

    try:
        with patch.object(
            coresys.hardware.disk, "disk_usage_for_mount", side_effect=_fails_slowly
        ):
            waiter = asyncio.create_task(api_host._mount_usage(mount, 0))

            # Let the probe get going, then abandon it the way a disconnecting
            # client does
            await asyncio.sleep(0.05)
            waiter.cancel()
            with suppress(asyncio.CancelledError):
                await waiter

            # Give the orphaned probe time to fail, then force collection so an
            # unretrieved exception would be surfaced
            await asyncio.sleep(0.5)
            gc.collect()
            await asyncio.sleep(0)
    finally:
        loop.set_exception_handler(previous_handler)

    # Nothing at all should reach the loop exception handler: an unreachable
    # mount is an expected outcome, not a supervisor fault
    assert not reported
    # The entry is gone either way, so a later request starts a fresh probe
    assert not api_host._mount_usage_probes


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_disk_usage_api_default_wins_over_mount_of_that_name(
    api_client_with_prefix: tuple[TestClient, str], coresys: CoreSys
):
    """Test "default" always means the system disk.

    The mount name pattern permits a mount called "default", so the resolution
    order has to be documented and pinned: the reserved target wins.
    """
    api_client, prefix = api_client_with_prefix
    _register_mount(coresys, "default", UnitActiveState.ACTIVE)

    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 = []

        resp = await api_client.get(f"{prefix}/host/disks/default/usage")

    assert resp.status == 200
    result = await resp.json()
    # The system disk, not the mount
    assert result["data"]["id"] == "root"
    assert result["data"]["label"] == "Root"